addwin/chat
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: addwin:f9412857f5566001b9dea0700fc3052e86a544d8
Branches Tags
addwin:master
..
compare: addwin:d259621a5bf6c847f1918deb842aa1fe8543ea08
Branches Tags
addwin:master

No commits in common. "f9412857f5566001b9dea0700fc3052e86a544d8" and "d259621a5bf6c847f1918deb842aa1fe8543ea08" have entirely different histories.
f9412857f5 ... d259621a5b

24 changed files with 3 additions and 2023 deletions
Show stats Expand all files Collapse all files

6
.env.example
View file

@ -12,16 +12,12 @@ OAUTH2_PROVIDER_BASE_PUBLIC_URL=https://accounts.example.com/oauth2
OAUTH2_PROVIDER_BASE_PRIVATE_URL=https://accounts.example.com/oauth2 OAUTH2_PROVIDER_BASE_PRIVATE_URL=https://accounts.example.com/oauth2
OAUTH2_PROVIDER_CLIENT_ID= OAUTH2_PROVIDER_CLIENT_ID=
OAUTH2_PROVIDER_CLIENT_SECRET= OAUTH2_PROVIDER_CLIENT_SECRET=
# Must include accounts.profile:retrieve for the accounts client (see ACCOUNTS_BASE_PUBLIC_URL) # Must include accounts.profile:retrieve for the accounts client (see ACCOUNTS_BASE_PUBLIC_URL) to read user profiles
# and notifications.application.push:send for the notifications client (see NOTIFICATIONS_BASE_PUBLIC_URL)
OAUTH2_PROVIDER_SCOPES= OAUTH2_PROVIDER_SCOPES=
# Gooyal Accounts service — used to fetch public user profiles (name, avatar) for chat responses # Gooyal Accounts service — used to fetch public user profiles (name, avatar) for chat responses
ACCOUNTS_BASE_PUBLIC_URL=https://accounts.example.com ACCOUNTS_BASE_PUBLIC_URL=https://accounts.example.com
# Gooyal Notifications service — used to push a notification to the other participant(s) when a message is sent
NOTIFICATIONS_BASE_PUBLIC_URL=https://notifications.example.com
# Redis (used for cache, channel layer, and long-poll) # Redis (used for cache, channel layer, and long-poll)
REDIS_BASE_URL=redis://127.0.0.1:6379/1 REDIS_BASE_URL=redis://127.0.0.1:6379/1

39
apps/chat/events/publishers/push.py
View file

@ -1,42 +1,7 @@
import logging
from apps.chat.events.event import MessageSentEvent from apps.chat.events.event import MessageSentEvent
from apps.chat.models import ConversationParticipant
from apps.chat.services.account import AccountService
from utils.clients import notifications_client
logger = logging.getLogger(__name__)
MEDIA_MESSAGE_PREVIEWS = {
"image": "Sent a photo",
"video": "Sent a video",
"voice": "Sent a voice message",
}
class PushPublisher: class PushPublisher:
def __init__(self, account_service: AccountService | None = None, push_user=None): # Future: FCM / APNS integration
self._account = account_service or AccountService()
self._push_user = push_user or notifications_client.push_user
def publish(self, event: MessageSentEvent) -> None: def publish(self, event: MessageSentEvent) -> None:
recipient_uuids = list( pass
ConversationParticipant.objects.filter(conversation_id=event.chat_uuid)
.exclude(user_uuid=event.sender_uuid)
.values_list("user_uuid", flat=True)
)
if not recipient_uuids:
return
sender_info = self._account.get_public_info([event.sender_uuid])
title = sender_info.get(str(event.sender_uuid), {}).get("name") or "New message"
body = event.payload.get("text") or MEDIA_MESSAGE_PREVIEWS.get(
event.message_type, "Sent a message"
)
extras = {"conversation_uuid": str(event.chat_uuid), "post_id": event.post_id}
for recipient_uuid in recipient_uuids:
try:
self._push_user(recipient_uuid, title=title, message=body, extras=extras)
except Exception:
logger.exception("Failed to push notification to user %s", recipient_uuid)

119
apps/chat/tests/test_push_publisher.py
View file

@ -1,119 +0,0 @@
import uuid
from unittest.mock import Mock
import pytest
from apps.chat.events.event import MessageSentEvent
from apps.chat.events.publishers.push import PushPublisher
from apps.chat.models import Conversation, ConversationParticipant
def _make_publisher(*, public_info=None, push_user=None):
account_service = Mock()
account_service.get_public_info.return_value = public_info or {}
return (
PushPublisher(account_service=account_service, push_user=push_user or Mock()),
account_service,
)
@pytest.mark.django_db
def test_push_notifies_other_participants_with_sender_name():
conv = Conversation.objects.create(mattermost_channel_id="ch-push")
sender_uuid = uuid.uuid4()
recipient_uuid = uuid.uuid4()
ConversationParticipant.objects.create(conversation=conv, user_uuid=sender_uuid)
ConversationParticipant.objects.create(conversation=conv, user_uuid=recipient_uuid)
push_user = Mock()
publisher, account_service = _make_publisher(
public_info={str(sender_uuid): {"name": "Alice"}}, push_user=push_user
)
event = MessageSentEvent(
chat_uuid=conv.uuid,
post_id="post-1",
sender_uuid=sender_uuid,
message_type="text",
payload={"text": "Hello there", "object_key": None, "url": None},
)
publisher.publish(event)
account_service.get_public_info.assert_called_once_with([sender_uuid])
push_user.assert_called_once_with(
recipient_uuid,
title="Alice",
message="Hello there",
extras={"conversation_uuid": str(conv.uuid), "post_id": "post-1"},
)
@pytest.mark.django_db
def test_push_skips_sender_and_uses_media_preview_when_no_text():
conv = Conversation.objects.create(mattermost_channel_id="ch-push-media")
sender_uuid = uuid.uuid4()
recipient_uuid = uuid.uuid4()
ConversationParticipant.objects.create(conversation=conv, user_uuid=sender_uuid)
ConversationParticipant.objects.create(conversation=conv, user_uuid=recipient_uuid)
push_user = Mock()
publisher, _ = _make_publisher(push_user=push_user)
event = MessageSentEvent(
chat_uuid=conv.uuid,
post_id="post-2",
sender_uuid=sender_uuid,
message_type="image",
payload={"text": None, "object_key": "images/x.jpg", "url": "http://x"},
)
publisher.publish(event)
push_user.assert_called_once()
called_uuid, kwargs = push_user.call_args[0][0], push_user.call_args[1]
assert called_uuid == recipient_uuid
assert kwargs["title"] == "New message"
assert kwargs["message"] == "Sent a photo"
@pytest.mark.django_db
def test_push_noop_when_no_other_participants():
conv = Conversation.objects.create(mattermost_channel_id="ch-push-solo")
sender_uuid = uuid.uuid4()
ConversationParticipant.objects.create(conversation=conv, user_uuid=sender_uuid)
push_user = Mock()
publisher, account_service = _make_publisher(push_user=push_user)
event = MessageSentEvent(
chat_uuid=conv.uuid,
post_id="post-3",
sender_uuid=sender_uuid,
message_type="text",
payload={"text": "Hi", "object_key": None, "url": None},
)
publisher.publish(event)
push_user.assert_not_called()
account_service.get_public_info.assert_not_called()
@pytest.mark.django_db
def test_push_swallows_per_recipient_failures():
conv = Conversation.objects.create(mattermost_channel_id="ch-push-fail")
sender_uuid = uuid.uuid4()
recipient_uuid = uuid.uuid4()
ConversationParticipant.objects.create(conversation=conv, user_uuid=sender_uuid)
ConversationParticipant.objects.create(conversation=conv, user_uuid=recipient_uuid)
push_user = Mock(side_effect=Exception("notifications service down"))
publisher, _ = _make_publisher(push_user=push_user)
event = MessageSentEvent(
chat_uuid=conv.uuid,
post_id="post-4",
sender_uuid=sender_uuid,
message_type="text",
payload={"text": "Hi", "object_key": None, "url": None},
)
publisher.publish(event) # must not raise

1
env.sample
View file

@ -10,7 +10,6 @@ OAUTH2_PROVIDER_CLIENT_ID=QkeJuvsPvwV2wjO2InQ7AIKfhCr0HqBWNtit2EHj
OAUTH2_PROVIDER_CLIENT_SECRET=S9AYajQRffgI7SzsxUmpzY7IMkVqlux8sW0A1scoBKgNfUFjXk3uwIMGbkBF9erMYONNVbSWemgODo6ArCDlg4UXYhaSZ8zolLhagfJ5Fk92RmrMLyY2rKv4cb209tPw OAUTH2_PROVIDER_CLIENT_SECRET=S9AYajQRffgI7SzsxUmpzY7IMkVqlux8sW0A1scoBKgNfUFjXk3uwIMGbkBF9erMYONNVbSWemgODo6ArCDlg4UXYhaSZ8zolLhagfJ5Fk92RmrMLyY2rKv4cb209tPw
OAUTH2_PROVIDER_SCOPES=accounts.profile:retrieve wallet.application.deposit:verify wallet.application.deposit:submit OAUTH2_PROVIDER_SCOPES=accounts.profile:retrieve wallet.application.deposit:verify wallet.application.deposit:submit
ACCOUNTS_BASE_PUBLIC_URL=https://accounts-staging.gooyal.ir ACCOUNTS_BASE_PUBLIC_URL=https://accounts-staging.gooyal.ir
NOTIFICATIONS_BASE_PUBLIC_URL=https://notifications-staging.gooyal.ir
REDIS_BASE_URL=redis://127.0.0.1:6379/1 REDIS_BASE_URL=redis://127.0.0.1:6379/1
MATTERMOST_TOKEN=3fktq5yzw7yp7m4ty7d3nx8gph MATTERMOST_TOKEN=3fktq5yzw7yp7m4ty7d3nx8gph

8
utils/clients/gooyal_notifications_client/__init__.py
View file

@ -1,8 +0,0 @@
"""A client library for accessing Notifications"""
from .client import AuthenticatedClient, Client
__all__ = (
"AuthenticatedClient",
"Client",
)

1
utils/clients/gooyal_notifications_client/api/__init__.py
View file

@ -1 +0,0 @@
"""Contains methods for accessing the API"""

1
utils/clients/gooyal_notifications_client/api/email/__init__.py
View file

@ -1 +0,0 @@
"""Contains endpoint functions for accessing the API"""

164
utils/clients/gooyal_notifications_client/api/email/email_application_application_create.py
View file

@ -1,164 +0,0 @@
from http import HTTPStatus
from typing import Any, Optional, Union
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.email import Email
from ...types import Response
def _get_kwargs(
user_uuid: str,
*,
body: Email,
) -> dict[str, Any]:
headers: dict[str, Any] = {}
_kwargs: dict[str, Any] = {
"method": "post",
"url": f"/email/application/{user_uuid}/application/",
}
_kwargs["json"] = body.to_dict()
headers["Content-Type"] = "application/json"
_kwargs["headers"] = headers
return _kwargs
def _parse_response(*, client: Union[AuthenticatedClient, Client], response: httpx.Response) -> Optional[Email]:
if response.status_code == 201:
response_201 = Email.from_dict(response.json())
return response_201
if client.raise_on_unexpected_status:
raise errors.UnexpectedStatus(response.status_code, response.content)
else:
return None
def _build_response(*, client: Union[AuthenticatedClient, Client], response: httpx.Response) -> Response[Email]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)
def sync_detailed(
user_uuid: str,
*,
client: AuthenticatedClient,
body: Email,
) -> Response[Email]:
"""
Args:
user_uuid (str):
body (Email):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[Email]
"""
kwargs = _get_kwargs(
user_uuid=user_uuid,
body=body,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
user_uuid: str,
*,
client: AuthenticatedClient,
body: Email,
) -> Optional[Email]:
"""
Args:
user_uuid (str):
body (Email):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Email
"""
return sync_detailed(
user_uuid=user_uuid,
client=client,
body=body,
).parsed
async def asyncio_detailed(
user_uuid: str,
*,
client: AuthenticatedClient,
body: Email,
) -> Response[Email]:
"""
Args:
user_uuid (str):
body (Email):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[Email]
"""
kwargs = _get_kwargs(
user_uuid=user_uuid,
body=body,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
user_uuid: str,
*,
client: AuthenticatedClient,
body: Email,
) -> Optional[Email]:
"""
Args:
user_uuid (str):
body (Email):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Email
"""
return (
await asyncio_detailed(
user_uuid=user_uuid,
client=client,
body=body,
)
).parsed

183
utils/clients/gooyal_notifications_client/api/email/email_application_application_list.py
View file

@ -1,183 +0,0 @@
from http import HTTPStatus
from typing import Any, Optional, Union
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.paginated_email_list import PaginatedEmailList
from ...types import UNSET, Response, Unset
def _get_kwargs(
user_uuid: str,
*,
limit: Union[Unset, int] = UNSET,
offset: Union[Unset, int] = UNSET,
) -> dict[str, Any]:
params: dict[str, Any] = {}
params["limit"] = limit
params["offset"] = offset
params = {k: v for k, v in params.items() if v is not UNSET and v is not None}
_kwargs: dict[str, Any] = {
"method": "get",
"url": f"/email/application/{user_uuid}/application/",
"params": params,
}
return _kwargs
def _parse_response(
*, client: Union[AuthenticatedClient, Client], response: httpx.Response
) -> Optional[PaginatedEmailList]:
if response.status_code == 200:
response_200 = PaginatedEmailList.from_dict(response.json())
return response_200
if client.raise_on_unexpected_status:
raise errors.UnexpectedStatus(response.status_code, response.content)
else:
return None
def _build_response(
*, client: Union[AuthenticatedClient, Client], response: httpx.Response
) -> Response[PaginatedEmailList]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)
def sync_detailed(
user_uuid: str,
*,
client: AuthenticatedClient,
limit: Union[Unset, int] = UNSET,
offset: Union[Unset, int] = UNSET,
) -> Response[PaginatedEmailList]:
"""
Args:
user_uuid (str):
limit (Union[Unset, int]):
offset (Union[Unset, int]):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[PaginatedEmailList]
"""
kwargs = _get_kwargs(
user_uuid=user_uuid,
limit=limit,
offset=offset,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
user_uuid: str,
*,
client: AuthenticatedClient,
limit: Union[Unset, int] = UNSET,
offset: Union[Unset, int] = UNSET,
) -> Optional[PaginatedEmailList]:
"""
Args:
user_uuid (str):
limit (Union[Unset, int]):
offset (Union[Unset, int]):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
PaginatedEmailList
"""
return sync_detailed(
user_uuid=user_uuid,
client=client,
limit=limit,
offset=offset,
).parsed
async def asyncio_detailed(
user_uuid: str,
*,
client: AuthenticatedClient,
limit: Union[Unset, int] = UNSET,
offset: Union[Unset, int] = UNSET,
) -> Response[PaginatedEmailList]:
"""
Args:
user_uuid (str):
limit (Union[Unset, int]):
offset (Union[Unset, int]):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[PaginatedEmailList]
"""
kwargs = _get_kwargs(
user_uuid=user_uuid,
limit=limit,
offset=offset,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
user_uuid: str,
*,
client: AuthenticatedClient,
limit: Union[Unset, int] = UNSET,
offset: Union[Unset, int] = UNSET,
) -> Optional[PaginatedEmailList]:
"""
Args:
user_uuid (str):
limit (Union[Unset, int]):
offset (Union[Unset, int]):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
PaginatedEmailList
"""
return (
await asyncio_detailed(
user_uuid=user_uuid,
client=client,
limit=limit,
offset=offset,
)
).parsed

1
utils/clients/gooyal_notifications_client/api/push/__init__.py
View file

@ -1 +0,0 @@
"""Contains endpoint functions for accessing the API"""

164
utils/clients/gooyal_notifications_client/api/push/push_application_application_create.py
View file

@ -1,164 +0,0 @@
from http import HTTPStatus
from typing import Any, Optional, Union
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.push_message import PushMessage
from ...types import Response
def _get_kwargs(
user_uuid: str,
*,
body: PushMessage,
) -> dict[str, Any]:
headers: dict[str, Any] = {}
_kwargs: dict[str, Any] = {
"method": "post",
"url": f"/push/application/{user_uuid}/application/",
}
_kwargs["json"] = body.to_dict()
headers["Content-Type"] = "application/json"
_kwargs["headers"] = headers
return _kwargs
def _parse_response(*, client: Union[AuthenticatedClient, Client], response: httpx.Response) -> Optional[PushMessage]:
if response.status_code == 201:
response_201 = PushMessage.from_dict(response.json())
return response_201
if client.raise_on_unexpected_status:
raise errors.UnexpectedStatus(response.status_code, response.content)
else:
return None
def _build_response(*, client: Union[AuthenticatedClient, Client], response: httpx.Response) -> Response[PushMessage]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)
def sync_detailed(
user_uuid: str,
*,
client: AuthenticatedClient,
body: PushMessage,
) -> Response[PushMessage]:
"""
Args:
user_uuid (str):
body (PushMessage):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[PushMessage]
"""
kwargs = _get_kwargs(
user_uuid=user_uuid,
body=body,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
user_uuid: str,
*,
client: AuthenticatedClient,
body: PushMessage,
) -> Optional[PushMessage]:
"""
Args:
user_uuid (str):
body (PushMessage):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
PushMessage
"""
return sync_detailed(
user_uuid=user_uuid,
client=client,
body=body,
).parsed
async def asyncio_detailed(
user_uuid: str,
*,
client: AuthenticatedClient,
body: PushMessage,
) -> Response[PushMessage]:
"""
Args:
user_uuid (str):
body (PushMessage):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[PushMessage]
"""
kwargs = _get_kwargs(
user_uuid=user_uuid,
body=body,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
user_uuid: str,
*,
client: AuthenticatedClient,
body: PushMessage,
) -> Optional[PushMessage]:
"""
Args:
user_uuid (str):
body (PushMessage):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
PushMessage
"""
return (
await asyncio_detailed(
user_uuid=user_uuid,
client=client,
body=body,
)
).parsed

183
utils/clients/gooyal_notifications_client/api/push/push_application_application_list.py
View file

@ -1,183 +0,0 @@
from http import HTTPStatus
from typing import Any, Optional, Union
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.paginated_push_message_list import PaginatedPushMessageList
from ...types import UNSET, Response, Unset
def _get_kwargs(
user_uuid: str,
*,
limit: Union[Unset, int] = UNSET,
offset: Union[Unset, int] = UNSET,
) -> dict[str, Any]:
params: dict[str, Any] = {}
params["limit"] = limit
params["offset"] = offset
params = {k: v for k, v in params.items() if v is not UNSET and v is not None}
_kwargs: dict[str, Any] = {
"method": "get",
"url": f"/push/application/{user_uuid}/application/",
"params": params,
}
return _kwargs
def _parse_response(
*, client: Union[AuthenticatedClient, Client], response: httpx.Response
) -> Optional[PaginatedPushMessageList]:
if response.status_code == 200:
response_200 = PaginatedPushMessageList.from_dict(response.json())
return response_200
if client.raise_on_unexpected_status:
raise errors.UnexpectedStatus(response.status_code, response.content)
else:
return None
def _build_response(
*, client: Union[AuthenticatedClient, Client], response: httpx.Response
) -> Response[PaginatedPushMessageList]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)
def sync_detailed(
user_uuid: str,
*,
client: AuthenticatedClient,
limit: Union[Unset, int] = UNSET,
offset: Union[Unset, int] = UNSET,
) -> Response[PaginatedPushMessageList]:
"""
Args:
user_uuid (str):
limit (Union[Unset, int]):
offset (Union[Unset, int]):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[PaginatedPushMessageList]
"""
kwargs = _get_kwargs(
user_uuid=user_uuid,
limit=limit,
offset=offset,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
user_uuid: str,
*,
client: AuthenticatedClient,
limit: Union[Unset, int] = UNSET,
offset: Union[Unset, int] = UNSET,
) -> Optional[PaginatedPushMessageList]:
"""
Args:
user_uuid (str):
limit (Union[Unset, int]):
offset (Union[Unset, int]):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
PaginatedPushMessageList
"""
return sync_detailed(
user_uuid=user_uuid,
client=client,
limit=limit,
offset=offset,
).parsed
async def asyncio_detailed(
user_uuid: str,
*,
client: AuthenticatedClient,
limit: Union[Unset, int] = UNSET,
offset: Union[Unset, int] = UNSET,
) -> Response[PaginatedPushMessageList]:
"""
Args:
user_uuid (str):
limit (Union[Unset, int]):
offset (Union[Unset, int]):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[PaginatedPushMessageList]
"""
kwargs = _get_kwargs(
user_uuid=user_uuid,
limit=limit,
offset=offset,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
user_uuid: str,
*,
client: AuthenticatedClient,
limit: Union[Unset, int] = UNSET,
offset: Union[Unset, int] = UNSET,
) -> Optional[PaginatedPushMessageList]:
"""
Args:
user_uuid (str):
limit (Union[Unset, int]):
offset (Union[Unset, int]):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
PaginatedPushMessageList
"""
return (
await asyncio_detailed(
user_uuid=user_uuid,
client=client,
limit=limit,
offset=offset,
)
).parsed

119
utils/clients/gooyal_notifications_client/api/push/push_user_push_user_retrieve.py
View file

@ -1,119 +0,0 @@
from http import HTTPStatus
from typing import Any, Optional, Union
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.push_user import PushUser
from ...types import Response
def _get_kwargs() -> dict[str, Any]:
_kwargs: dict[str, Any] = {
"method": "get",
"url": "/push/user/push_user/",
}
return _kwargs
def _parse_response(*, client: Union[AuthenticatedClient, Client], response: httpx.Response) -> Optional[PushUser]:
if response.status_code == 200:
response_200 = PushUser.from_dict(response.json())
return response_200
if client.raise_on_unexpected_status:
raise errors.UnexpectedStatus(response.status_code, response.content)
else:
return None
def _build_response(*, client: Union[AuthenticatedClient, Client], response: httpx.Response) -> Response[PushUser]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)
def sync_detailed(
*,
client: AuthenticatedClient,
) -> Response[PushUser]:
"""
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[PushUser]
"""
kwargs = _get_kwargs()
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
*,
client: AuthenticatedClient,
) -> Optional[PushUser]:
"""
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
PushUser
"""
return sync_detailed(
client=client,
).parsed
async def asyncio_detailed(
*,
client: AuthenticatedClient,
) -> Response[PushUser]:
"""
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[PushUser]
"""
kwargs = _get_kwargs()
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
*,
client: AuthenticatedClient,
) -> Optional[PushUser]:
"""
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
PushUser
"""
return (
await asyncio_detailed(
client=client,
)
).parsed

268
utils/clients/gooyal_notifications_client/client.py
View file

@ -1,268 +0,0 @@
import ssl
from typing import Any, Optional, Union
import httpx
from attrs import define, evolve, field
@define
class Client:
"""A class for keeping track of data related to the API
The following are accepted as keyword arguments and will be used to construct httpx Clients internally:
``base_url``: The base URL for the API, all requests are made to a relative path to this URL
``cookies``: A dictionary of cookies to be sent with every request
``headers``: A dictionary of headers to be sent with every request
``timeout``: The maximum amount of a time a request can take. API functions will raise
httpx.TimeoutException if this is exceeded.
``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production,
but can be set to False for testing purposes.
``follow_redirects``: Whether or not to follow redirects. Default value is False.
``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor.
Attributes:
raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a
status code that was not documented in the source OpenAPI document. Can also be provided as a keyword
argument to the constructor.
"""
raise_on_unexpected_status: bool = field(default=False, kw_only=True)
_base_url: str = field(alias="base_url")
_cookies: dict[str, str] = field(factory=dict, kw_only=True, alias="cookies")
_headers: dict[str, str] = field(factory=dict, kw_only=True, alias="headers")
_timeout: Optional[httpx.Timeout] = field(default=None, kw_only=True, alias="timeout")
_verify_ssl: Union[str, bool, ssl.SSLContext] = field(default=True, kw_only=True, alias="verify_ssl")
_follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects")
_httpx_args: dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args")
_client: Optional[httpx.Client] = field(default=None, init=False)
_async_client: Optional[httpx.AsyncClient] = field(default=None, init=False)
def with_headers(self, headers: dict[str, str]) -> "Client":
"""Get a new client matching this one with additional headers"""
if self._client is not None:
self._client.headers.update(headers)
if self._async_client is not None:
self._async_client.headers.update(headers)
return evolve(self, headers={**self._headers, **headers})
def with_cookies(self, cookies: dict[str, str]) -> "Client":
"""Get a new client matching this one with additional cookies"""
if self._client is not None:
self._client.cookies.update(cookies)
if self._async_client is not None:
self._async_client.cookies.update(cookies)
return evolve(self, cookies={**self._cookies, **cookies})
def with_timeout(self, timeout: httpx.Timeout) -> "Client":
"""Get a new client matching this one with a new timeout (in seconds)"""
if self._client is not None:
self._client.timeout = timeout
if self._async_client is not None:
self._async_client.timeout = timeout
return evolve(self, timeout=timeout)
def set_httpx_client(self, client: httpx.Client) -> "Client":
"""Manually set the underlying httpx.Client
**NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
"""
self._client = client
return self
def get_httpx_client(self) -> httpx.Client:
"""Get the underlying httpx.Client, constructing a new one if not previously set"""
if self._client is None:
self._client = httpx.Client(
base_url=self._base_url,
cookies=self._cookies,
headers=self._headers,
timeout=self._timeout,
verify=self._verify_ssl,
follow_redirects=self._follow_redirects,
**self._httpx_args,
)
return self._client
def __enter__(self) -> "Client":
"""Enter a context manager for self.client—you cannot enter twice (see httpx docs)"""
self.get_httpx_client().__enter__()
return self
def __exit__(self, *args: Any, **kwargs: Any) -> None:
"""Exit a context manager for internal httpx.Client (see httpx docs)"""
self.get_httpx_client().__exit__(*args, **kwargs)
def set_async_httpx_client(self, async_client: httpx.AsyncClient) -> "Client":
"""Manually the underlying httpx.AsyncClient
**NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
"""
self._async_client = async_client
return self
def get_async_httpx_client(self) -> httpx.AsyncClient:
"""Get the underlying httpx.AsyncClient, constructing a new one if not previously set"""
if self._async_client is None:
self._async_client = httpx.AsyncClient(
base_url=self._base_url,
cookies=self._cookies,
headers=self._headers,
timeout=self._timeout,
verify=self._verify_ssl,
follow_redirects=self._follow_redirects,
**self._httpx_args,
)
return self._async_client
async def __aenter__(self) -> "Client":
"""Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)"""
await self.get_async_httpx_client().__aenter__()
return self
async def __aexit__(self, *args: Any, **kwargs: Any) -> None:
"""Exit a context manager for underlying httpx.AsyncClient (see httpx docs)"""
await self.get_async_httpx_client().__aexit__(*args, **kwargs)
@define
class AuthenticatedClient:
"""A Client which has been authenticated for use on secured endpoints
The following are accepted as keyword arguments and will be used to construct httpx Clients internally:
``base_url``: The base URL for the API, all requests are made to a relative path to this URL
``cookies``: A dictionary of cookies to be sent with every request
``headers``: A dictionary of headers to be sent with every request
``timeout``: The maximum amount of a time a request can take. API functions will raise
httpx.TimeoutException if this is exceeded.
``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production,
but can be set to False for testing purposes.
``follow_redirects``: Whether or not to follow redirects. Default value is False.
``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor.
Attributes:
raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a
status code that was not documented in the source OpenAPI document. Can also be provided as a keyword
argument to the constructor.
token: The token to use for authentication
prefix: The prefix to use for the Authorization header
auth_header_name: The name of the Authorization header
"""
raise_on_unexpected_status: bool = field(default=False, kw_only=True)
_base_url: str = field(alias="base_url")
_cookies: dict[str, str] = field(factory=dict, kw_only=True, alias="cookies")
_headers: dict[str, str] = field(factory=dict, kw_only=True, alias="headers")
_timeout: Optional[httpx.Timeout] = field(default=None, kw_only=True, alias="timeout")
_verify_ssl: Union[str, bool, ssl.SSLContext] = field(default=True, kw_only=True, alias="verify_ssl")
_follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects")
_httpx_args: dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args")
_client: Optional[httpx.Client] = field(default=None, init=False)
_async_client: Optional[httpx.AsyncClient] = field(default=None, init=False)
token: str
prefix: str = "Bearer"
auth_header_name: str = "Authorization"
def with_headers(self, headers: dict[str, str]) -> "AuthenticatedClient":
"""Get a new client matching this one with additional headers"""
if self._client is not None:
self._client.headers.update(headers)
if self._async_client is not None:
self._async_client.headers.update(headers)
return evolve(self, headers={**self._headers, **headers})
def with_cookies(self, cookies: dict[str, str]) -> "AuthenticatedClient":
"""Get a new client matching this one with additional cookies"""
if self._client is not None:
self._client.cookies.update(cookies)
if self._async_client is not None:
self._async_client.cookies.update(cookies)
return evolve(self, cookies={**self._cookies, **cookies})
def with_timeout(self, timeout: httpx.Timeout) -> "AuthenticatedClient":
"""Get a new client matching this one with a new timeout (in seconds)"""
if self._client is not None:
self._client.timeout = timeout
if self._async_client is not None:
self._async_client.timeout = timeout
return evolve(self, timeout=timeout)
def set_httpx_client(self, client: httpx.Client) -> "AuthenticatedClient":
"""Manually set the underlying httpx.Client
**NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
"""
self._client = client
return self
def get_httpx_client(self) -> httpx.Client:
"""Get the underlying httpx.Client, constructing a new one if not previously set"""
if self._client is None:
self._headers[self.auth_header_name] = f"{self.prefix} {self.token}" if self.prefix else self.token
self._client = httpx.Client(
base_url=self._base_url,
cookies=self._cookies,
headers=self._headers,
timeout=self._timeout,
verify=self._verify_ssl,
follow_redirects=self._follow_redirects,
**self._httpx_args,
)
return self._client
def __enter__(self) -> "AuthenticatedClient":
"""Enter a context manager for self.client—you cannot enter twice (see httpx docs)"""
self.get_httpx_client().__enter__()
return self
def __exit__(self, *args: Any, **kwargs: Any) -> None:
"""Exit a context manager for internal httpx.Client (see httpx docs)"""
self.get_httpx_client().__exit__(*args, **kwargs)
def set_async_httpx_client(self, async_client: httpx.AsyncClient) -> "AuthenticatedClient":
"""Manually the underlying httpx.AsyncClient
**NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
"""
self._async_client = async_client
return self
def get_async_httpx_client(self) -> httpx.AsyncClient:
"""Get the underlying httpx.AsyncClient, constructing a new one if not previously set"""
if self._async_client is None:
self._headers[self.auth_header_name] = f"{self.prefix} {self.token}" if self.prefix else self.token
self._async_client = httpx.AsyncClient(
base_url=self._base_url,
cookies=self._cookies,
headers=self._headers,
timeout=self._timeout,
verify=self._verify_ssl,
follow_redirects=self._follow_redirects,
**self._httpx_args,
)
return self._async_client
async def __aenter__(self) -> "AuthenticatedClient":
"""Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)"""
await self.get_async_httpx_client().__aenter__()
return self
async def __aexit__(self, *args: Any, **kwargs: Any) -> None:
"""Exit a context manager for underlying httpx.AsyncClient (see httpx docs)"""
await self.get_async_httpx_client().__aexit__(*args, **kwargs)

16
utils/clients/gooyal_notifications_client/errors.py
View file

@ -1,16 +0,0 @@
"""Contains shared errors types that can be raised from API functions"""
class UnexpectedStatus(Exception):
"""Raised by api functions when the response status an undocumented status and Client.raise_on_unexpected_status is True"""
def __init__(self, status_code: int, content: bytes):
self.status_code = status_code
self.content = content
super().__init__(
f"Unexpected status code: {status_code}\n\nResponse content:\n{content.decode(errors='ignore')}"
)
__all__ = ["UnexpectedStatus"]

15
utils/clients/gooyal_notifications_client/models/__init__.py
View file

@ -1,15 +0,0 @@
"""Contains all the data models used in inputs/outputs"""
from .email import Email
from .paginated_email_list import PaginatedEmailList
from .paginated_push_message_list import PaginatedPushMessageList
from .push_message import PushMessage
from .push_user import PushUser
__all__ = (
"Email",
"PaginatedEmailList",
"PaginatedPushMessageList",
"PushMessage",
"PushUser",
)

133
utils/clients/gooyal_notifications_client/models/email.py
View file

@ -1,133 +0,0 @@
from collections.abc import Mapping
from typing import Any, TypeVar, Union, cast
from uuid import UUID
from attrs import define as _attrs_define
from attrs import field as _attrs_field
from ..types import UNSET, Unset
T = TypeVar("T", bound="Email")
@_attrs_define
class Email:
"""
Attributes:
title (str):
message (str):
user (UUID):
application (Union[None, UUID]):
queue (Union[None, UUID, Unset]):
"""
title: str
message: str
user: UUID
application: Union[None, UUID]
queue: Union[None, UUID, Unset] = UNSET
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
def to_dict(self) -> dict[str, Any]:
title = self.title
message = self.message
user = str(self.user)
application: Union[None, str]
if isinstance(self.application, UUID):
application = str(self.application)
else:
application = self.application
queue: Union[None, Unset, str]
if isinstance(self.queue, Unset):
queue = UNSET
elif isinstance(self.queue, UUID):
queue = str(self.queue)
else:
queue = self.queue
field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"title": title,
"message": message,
"user": user,
"application": application,
}
)
if queue is not UNSET:
field_dict["queue"] = queue
return field_dict
@classmethod
def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
d = dict(src_dict)
title = d.pop("title")
message = d.pop("message")
user = UUID(d.pop("user"))
def _parse_application(data: object) -> Union[None, UUID]:
if data is None:
return data
try:
if not isinstance(data, str):
raise TypeError()
application_type_0 = UUID(data)
return application_type_0
except: # noqa: E722
pass
return cast(Union[None, UUID], data)
application = _parse_application(d.pop("application"))
def _parse_queue(data: object) -> Union[None, UUID, Unset]:
if data is None:
return data
if isinstance(data, Unset):
return data
try:
if not isinstance(data, str):
raise TypeError()
queue_type_0 = UUID(data)
return queue_type_0
except: # noqa: E722
pass
return cast(Union[None, UUID, Unset], data)
queue = _parse_queue(d.pop("queue", UNSET))
email = cls(
title=title,
message=message,
user=user,
application=application,
queue=queue,
)
email.additional_properties = d
return email
@property
def additional_keys(self) -> list[str]:
return list(self.additional_properties.keys())
def __getitem__(self, key: str) -> Any:
return self.additional_properties[key]
def __setitem__(self, key: str, value: Any) -> None:
self.additional_properties[key] = value
def __delitem__(self, key: str) -> None:
del self.additional_properties[key]
def __contains__(self, key: str) -> bool:
return key in self.additional_properties

123
utils/clients/gooyal_notifications_client/models/paginated_email_list.py
View file

@ -1,123 +0,0 @@
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, TypeVar, Union, cast
from attrs import define as _attrs_define
from attrs import field as _attrs_field
from ..types import UNSET, Unset
if TYPE_CHECKING:
from ..models.email import Email
T = TypeVar("T", bound="PaginatedEmailList")
@_attrs_define
class PaginatedEmailList:
"""
Attributes:
count (int): Example: 123.
results (list['Email']):
next_ (Union[None, Unset, str]): Example: http://api.example.org/accounts/?offset=400&limit=100.
previous (Union[None, Unset, str]): Example: http://api.example.org/accounts/?offset=200&limit=100.
"""
count: int
results: list["Email"]
next_: Union[None, Unset, str] = UNSET
previous: Union[None, Unset, str] = UNSET
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
def to_dict(self) -> dict[str, Any]:
count = self.count
results = []
for results_item_data in self.results:
results_item = results_item_data.to_dict()
results.append(results_item)
next_: Union[None, Unset, str]
if isinstance(self.next_, Unset):
next_ = UNSET
else:
next_ = self.next_
previous: Union[None, Unset, str]
if isinstance(self.previous, Unset):
previous = UNSET
else:
previous = self.previous
field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"count": count,
"results": results,
}
)
if next_ is not UNSET:
field_dict["next"] = next_
if previous is not UNSET:
field_dict["previous"] = previous
return field_dict
@classmethod
def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
from ..models.email import Email
d = dict(src_dict)
count = d.pop("count")
results = []
_results = d.pop("results")
for results_item_data in _results:
results_item = Email.from_dict(results_item_data)
results.append(results_item)
def _parse_next_(data: object) -> Union[None, Unset, str]:
if data is None:
return data
if isinstance(data, Unset):
return data
return cast(Union[None, Unset, str], data)
next_ = _parse_next_(d.pop("next", UNSET))
def _parse_previous(data: object) -> Union[None, Unset, str]:
if data is None:
return data
if isinstance(data, Unset):
return data
return cast(Union[None, Unset, str], data)
previous = _parse_previous(d.pop("previous", UNSET))
paginated_email_list = cls(
count=count,
results=results,
next_=next_,
previous=previous,
)
paginated_email_list.additional_properties = d
return paginated_email_list
@property
def additional_keys(self) -> list[str]:
return list(self.additional_properties.keys())
def __getitem__(self, key: str) -> Any:
return self.additional_properties[key]
def __setitem__(self, key: str, value: Any) -> None:
self.additional_properties[key] = value
def __delitem__(self, key: str) -> None:
del self.additional_properties[key]
def __contains__(self, key: str) -> bool:
return key in self.additional_properties

123
utils/clients/gooyal_notifications_client/models/paginated_push_message_list.py
View file

@ -1,123 +0,0 @@
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, TypeVar, Union, cast
from attrs import define as _attrs_define
from attrs import field as _attrs_field
from ..types import UNSET, Unset
if TYPE_CHECKING:
from ..models.push_message import PushMessage
T = TypeVar("T", bound="PaginatedPushMessageList")
@_attrs_define
class PaginatedPushMessageList:
"""
Attributes:
count (int): Example: 123.
results (list['PushMessage']):
next_ (Union[None, Unset, str]): Example: http://api.example.org/accounts/?offset=400&limit=100.
previous (Union[None, Unset, str]): Example: http://api.example.org/accounts/?offset=200&limit=100.
"""
count: int
results: list["PushMessage"]
next_: Union[None, Unset, str] = UNSET
previous: Union[None, Unset, str] = UNSET
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
def to_dict(self) -> dict[str, Any]:
count = self.count
results = []
for results_item_data in self.results:
results_item = results_item_data.to_dict()
results.append(results_item)
next_: Union[None, Unset, str]
if isinstance(self.next_, Unset):
next_ = UNSET
else:
next_ = self.next_
previous: Union[None, Unset, str]
if isinstance(self.previous, Unset):
previous = UNSET
else:
previous = self.previous
field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"count": count,
"results": results,
}
)
if next_ is not UNSET:
field_dict["next"] = next_
if previous is not UNSET:
field_dict["previous"] = previous
return field_dict
@classmethod
def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
from ..models.push_message import PushMessage
d = dict(src_dict)
count = d.pop("count")
results = []
_results = d.pop("results")
for results_item_data in _results:
results_item = PushMessage.from_dict(results_item_data)
results.append(results_item)
def _parse_next_(data: object) -> Union[None, Unset, str]:
if data is None:
return data
if isinstance(data, Unset):
return data
return cast(Union[None, Unset, str], data)
next_ = _parse_next_(d.pop("next", UNSET))
def _parse_previous(data: object) -> Union[None, Unset, str]:
if data is None:
return data
if isinstance(data, Unset):
return data
return cast(Union[None, Unset, str], data)
previous = _parse_previous(d.pop("previous", UNSET))
paginated_push_message_list = cls(
count=count,
results=results,
next_=next_,
previous=previous,
)
paginated_push_message_list.additional_properties = d
return paginated_push_message_list
@property
def additional_keys(self) -> list[str]:
return list(self.additional_properties.keys())
def __getitem__(self, key: str) -> Any:
return self.additional_properties[key]
def __setitem__(self, key: str, value: Any) -> None:
self.additional_properties[key] = value
def __delitem__(self, key: str) -> None:
del self.additional_properties[key]
def __contains__(self, key: str) -> bool:
return key in self.additional_properties

121
utils/clients/gooyal_notifications_client/models/push_message.py
View file

@ -1,121 +0,0 @@
from collections.abc import Mapping
from typing import Any, TypeVar, Union, cast
from uuid import UUID
from attrs import define as _attrs_define
from attrs import field as _attrs_field
from ..types import UNSET, Unset
T = TypeVar("T", bound="PushMessage")
@_attrs_define
class PushMessage:
"""
Attributes:
push_user (UUID):
application (Union[None, UUID]):
title (str):
message (str):
priority (Union[Unset, int]):
extras (Union[Unset, Any]):
"""
push_user: UUID
application: Union[None, UUID]
title: str
message: str
priority: Union[Unset, int] = UNSET
extras: Union[Unset, Any] = UNSET
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
def to_dict(self) -> dict[str, Any]:
push_user = str(self.push_user)
application: Union[None, str]
if isinstance(self.application, UUID):
application = str(self.application)
else:
application = self.application
title = self.title
message = self.message
priority = self.priority
extras = self.extras
field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"push_user": push_user,
"application": application,
"title": title,
"message": message,
}
)
if priority is not UNSET:
field_dict["priority"] = priority
if extras is not UNSET:
field_dict["extras"] = extras
return field_dict
@classmethod
def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
d = dict(src_dict)
push_user = UUID(d.pop("push_user"))
def _parse_application(data: object) -> Union[None, UUID]:
if data is None:
return data
try:
if not isinstance(data, str):
raise TypeError()
application_type_0 = UUID(data)
return application_type_0
except: # noqa: E722
pass
return cast(Union[None, UUID], data)
application = _parse_application(d.pop("application"))
title = d.pop("title")
message = d.pop("message")
priority = d.pop("priority", UNSET)
extras = d.pop("extras", UNSET)
push_message = cls(
push_user=push_user,
application=application,
title=title,
message=message,
priority=priority,
extras=extras,
)
push_message.additional_properties = d
return push_message
@property
def additional_keys(self) -> list[str]:
return list(self.additional_properties.keys())
def __getitem__(self, key: str) -> Any:
return self.additional_properties[key]
def __setitem__(self, key: str, value: Any) -> None:
self.additional_properties[key] = value
def __delitem__(self, key: str) -> None:
del self.additional_properties[key]
def __contains__(self, key: str) -> bool:
return key in self.additional_properties

102
utils/clients/gooyal_notifications_client/models/push_user.py
View file

@ -1,102 +0,0 @@
from collections.abc import Mapping
from typing import Any, TypeVar, Union, cast
from uuid import UUID
from attrs import define as _attrs_define
from attrs import field as _attrs_field
from ..types import UNSET, Unset
T = TypeVar("T", bound="PushUser")
@_attrs_define
class PushUser:
"""
Attributes:
user (UUID):
client_token (Union[None, Unset, str]):
application_token (Union[None, Unset, str]):
"""
user: UUID
client_token: Union[None, Unset, str] = UNSET
application_token: Union[None, Unset, str] = UNSET
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
def to_dict(self) -> dict[str, Any]:
user = str(self.user)
client_token: Union[None, Unset, str]
if isinstance(self.client_token, Unset):
client_token = UNSET
else:
client_token = self.client_token
application_token: Union[None, Unset, str]
if isinstance(self.application_token, Unset):
application_token = UNSET
else:
application_token = self.application_token
field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"user": user,
}
)
if client_token is not UNSET:
field_dict["client_token"] = client_token
if application_token is not UNSET:
field_dict["application_token"] = application_token
return field_dict
@classmethod
def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
d = dict(src_dict)
user = UUID(d.pop("user"))
def _parse_client_token(data: object) -> Union[None, Unset, str]:
if data is None:
return data
if isinstance(data, Unset):
return data
return cast(Union[None, Unset, str], data)
client_token = _parse_client_token(d.pop("client_token", UNSET))
def _parse_application_token(data: object) -> Union[None, Unset, str]:
if data is None:
return data
if isinstance(data, Unset):
return data
return cast(Union[None, Unset, str], data)
application_token = _parse_application_token(d.pop("application_token", UNSET))
push_user = cls(
user=user,
client_token=client_token,
application_token=application_token,
)
push_user.additional_properties = d
return push_user
@property
def additional_keys(self) -> list[str]:
return list(self.additional_properties.keys())
def __getitem__(self, key: str) -> Any:
return self.additional_properties[key]
def __setitem__(self, key: str, value: Any) -> None:
self.additional_properties[key] = value
def __delitem__(self, key: str) -> None:
del self.additional_properties[key]
def __contains__(self, key: str) -> bool:
return key in self.additional_properties

1
utils/clients/gooyal_notifications_client/py.typed
View file

@ -1 +0,0 @@
# Marker file for PEP 561

54
utils/clients/gooyal_notifications_client/types.py
View file

@ -1,54 +0,0 @@
"""Contains some shared types for properties"""
from collections.abc import Mapping, MutableMapping
from http import HTTPStatus
from typing import IO, BinaryIO, Generic, Literal, Optional, TypeVar, Union
from attrs import define
class Unset:
def __bool__(self) -> Literal[False]:
return False
UNSET: Unset = Unset()
# The types that `httpx.Client(files=)` can accept, copied from that library.
FileContent = Union[IO[bytes], bytes, str]
FileTypes = Union[
# (filename, file (or bytes), content_type)
tuple[Optional[str], FileContent, Optional[str]],
# (filename, file (or bytes), content_type, headers)
tuple[Optional[str], FileContent, Optional[str], Mapping[str, str]],
]
RequestFiles = list[tuple[str, FileTypes]]
@define
class File:
"""Contains information for file uploads"""
payload: BinaryIO
file_name: Optional[str] = None
mime_type: Optional[str] = None
def to_tuple(self) -> FileTypes:
"""Return a tuple representation that httpx will accept for multipart/form-data"""
return self.file_name, self.payload, self.mime_type
T = TypeVar("T")
@define
class Response(Generic[T]):
"""A response from an endpoint"""
status_code: HTTPStatus
content: bytes
headers: MutableMapping[str, str]
parsed: Optional[T]
__all__ = ["UNSET", "File", "FileTypes", "RequestFiles", "Response", "Unset"]

81
utils/clients/notifications_client.py
View file

@ -1,81 +0,0 @@
import time
import requests
from django.conf import settings
from django.core.cache import cache
from utils.clients.gooyal_notifications_client import AuthenticatedClient
from utils.clients.gooyal_notifications_client.api.push import push_application_application_create
ACCESS_TOKEN_CACHE_KEY = "chat_notifications_access_token"
def login_as_client_credentials() -> dict:
"""
Client-credentials OAuth2 grant against the gooyal accounts service,
reusing a cached access token until it's within 5s of expiry.
"""
cached = cache.get(ACCESS_TOKEN_CACHE_KEY)
if cached:
return cached
data = {
"grant_type": "client_credentials",
"scope": settings.OAUTH2_PROVIDER_SCOPES,
}
auth = (settings.OAUTH2_PROVIDER_CLIENT_ID, settings.OAUTH2_PROVIDER_CLIENT_SECRET)
response = requests.post(
f"{settings.OAUTH2_PROVIDER_BASE_PUBLIC_URL}/token/", data=data, auth=auth
)
response.raise_for_status()
auth_data = response.json()
if "access_token" in auth_data:
auth_data["created_at"] = time.time()
cache.set(
ACCESS_TOKEN_CACHE_KEY, auth_data, timeout=auth_data.get("expires_in", 300) - 5
)
return auth_data
def get_client() -> AuthenticatedClient:
access_token = login_as_client_credentials()["access_token"]
return AuthenticatedClient(
base_url=settings.NOTIFICATIONS_BASE_PUBLIC_URL, token=access_token
)
class _PushBody:
"""
Duck-typed request body: push_user is already in the URL and
application is resolved server-side from the OAuth2 client, so only
these fields are sent (mirrors the advertising service's client).
"""
def __init__(self, title: str, message: str, priority: int, extras):
self._title = title
self._message = message
self._priority = priority
self._extras = extras
def to_dict(self) -> dict:
return {
"title": self._title,
"message": self._message,
"priority": self._priority,
"extras": self._extras,
}
def push_user(user_uuid, title: str, message: str, priority: int = 5, extras=None):
"""
Send a push notification to a single user via the gooyal notifications
service. Requires the `notifications.application.push:send` scope on
this app's OAuth2 client credentials. Raises on failure — callers
decide whether a down notifications service should be swallowed.
"""
client = get_client()
body = _PushBody(title=title, message=message, priority=priority, extras=extras)
return push_application_application_create.sync(user_uuid=str(user_uuid), client=client, body=body)