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

Compare commits

base: addwin:fa3bca29df0102736ae7f3f6a10dfa06c2ad1786
Branches Tags
addwin:master
addwin:FEATURE/add-referral_proxy_api
addwin:FEATURE/support-accept-language-header
addwin:feat/send-auto-fill-sms
..
compare: addwin:e6ea563d197b5b653fdb033c07b3c08c1f1b29f9
Branches Tags
addwin:master
addwin:FEATURE/add-referral_proxy_api
addwin:FEATURE/support-accept-language-header
addwin:feat/send-auto-fill-sms

No commits in common. "fa3bca29df0102736ae7f3f6a10dfa06c2ad1786" and "e6ea563d197b5b653fdb033c07b3c08c1f1b29f9" have entirely different histories.
fa3bca29df ... e6ea563d19

34 changed files with 9 additions and 2940 deletions
Show stats Expand all files Collapse all files

20
apps/users/admin.py
View file

@ -1,5 +1,5 @@
from django.contrib import admin from django.contrib import admin
from .models import User, InviteRequest from .models import User
class UserAdmin(admin.ModelAdmin): class UserAdmin(admin.ModelAdmin):
@ -27,22 +27,4 @@ class UserAdmin(admin.ModelAdmin):
search_fields = ['pk', 'first_name', 'last_name', 'phone_number'] search_fields = ['pk', 'first_name', 'last_name', 'phone_number']
class InviteRequestAdmin(admin.ModelAdmin):
fields = [
'uuid',
'inviter',
'mobile_number',
'application',
'created_at',
'updated_at',
]
readonly_fields = ['uuid', 'created_at', 'updated_at']
list_display = ['uuid', 'mobile_number', 'inviter', 'application', 'created_at']
list_filter = ['created_at', 'application']
search_fields = ['uuid', 'mobile_number', 'inviter__uuid', 'inviter__phone_number']
raw_id_fields = ['inviter', 'application']
date_hierarchy = 'created_at'
admin.site.register(User, UserAdmin) admin.site.register(User, UserAdmin)
admin.site.register(InviteRequest, InviteRequestAdmin)

33
apps/users/migrations/0009_inviterequest.py
View file

@ -1,33 +0,0 @@
# Generated by Django 5.1.4 on 2026-07-22 13:45
import django.db.models.deletion
import uuid
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('users', '0008_alter_user_options'),
migrations.swappable_dependency(settings.OAUTH2_PROVIDER_APPLICATION_MODEL),
]
operations = [
migrations.CreateModel(
name='InviteRequest',
fields=[
('uuid', models.UUIDField(db_index=True, default=uuid.uuid4, primary_key=True, serialize=False, unique=True)),
('created_at', models.DateTimeField(auto_now_add=True, db_index=True)),
('updated_at', models.DateTimeField(auto_now=True, db_index=True)),
('mobile_number', models.CharField(db_index=True, max_length=30, verbose_name='mobile number')),
('application', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='invite_requests', to=settings.OAUTH2_PROVIDER_APPLICATION_MODEL)),
('inviter', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='invite_requests', to=settings.AUTH_USER_MODEL, verbose_name='inviter')),
],
options={
'verbose_name': 'invite request',
'verbose_name_plural': 'invite requests',
'ordering': ['-created_at'],
},
),
]

18
apps/users/migrations/0010_alter_inviterequest_mobile_number.py
View file

@ -1,18 +0,0 @@
# Generated by Django 5.1.4 on 2026-07-25
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('users', '0009_inviterequest'),
]
operations = [
migrations.AlterField(
model_name='inviterequest',
name='mobile_number',
field=models.CharField(db_index=True, max_length=30, unique=True, verbose_name='mobile number'),
),
]

27
apps/users/models.py
View file

@ -21,8 +21,6 @@ from .provinces_and_cities import state
from apps.users.constans import MAX_OTP_TRY, DEVELOPMENT_PHONE_NUMBERS, GenderChoices, EducationDegreeChoices, \ from apps.users.constans import MAX_OTP_TRY, DEVELOPMENT_PHONE_NUMBERS, GenderChoices, EducationDegreeChoices, \
MAX_OTP_VALID_DURATION MAX_OTP_VALID_DURATION
from oauth2_provider import settings as oauth2_settings
from utils.models import BaseModel
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -269,28 +267,3 @@ class User(AbstractUser):
def __str__(self): def __str__(self):
return self.name or self.get_full_name() or self.username or self.phone_number or self.email or _('no name') return self.name or self.get_full_name() or self.username or self.phone_number or self.email or _('no name')
class InviteRequest(BaseModel):
inviter = models.ForeignKey(
User,
on_delete=models.CASCADE,
related_name='invite_requests',
verbose_name=_('inviter'),
)
mobile_number = models.CharField(_('mobile number'), unique=True, max_length=30, db_index=True)
application = models.ForeignKey(
oauth2_settings.APPLICATION_MODEL,
on_delete=models.PROTECT,
related_name='invite_requests',
null=True,
blank=True
)
class Meta:
verbose_name = _('invite request')
verbose_name_plural = _('invite requests')
ordering = ['-created_at']
def __str__(self):
return f'{self.mobile_number} -> {self.inviter_id}'

45
apps/users/serializers.py
View file

@ -2,12 +2,12 @@ from django.core.validators import RegexValidator
from oauth2_provider.models import get_access_token_model from oauth2_provider.models import get_access_token_model
from rest_framework import serializers from rest_framework import serializers
from apps.users.models import User, InviteRequest from apps.users.models import User
from django.utils import timezone from django.utils import timezone
from django.utils.translation import gettext_lazy as _ from django.utils.translation import gettext_lazy as _
from unidecode import unidecode from unidecode import unidecode
from utils.exceptions import UnprocessableEntity, Conflict from utils.exceptions import UnprocessableEntity
phone_number_validator = RegexValidator(regex=r'^\+98[1-9]\d{9}$', message=_( phone_number_validator = RegexValidator(regex=r'^\+98[1-9]\d{9}$', message=_(
"Phone number must be entered in the format: '+989999999999'." "Phone number must be entered in the format: '+989999999999'."
@ -134,47 +134,6 @@ class ChangePasswordSerializer(serializers.Serializer):
new_password = serializers.CharField(required=True) new_password = serializers.CharField(required=True)
class InviteRequestSerializer(serializers.ModelSerializer):
invite_code = serializers.UUIDField(write_only=True)
class Meta:
model = InviteRequest
fields = ('uuid', 'invite_code', 'mobile_number', 'created_at')
read_only_fields = ('uuid', 'created_at')
def validate_invite_code(self, value):
if not User.objects.filter(uuid=value).exists():
raise serializers.ValidationError(_('Invalid invite code.'))
return value
def validate_mobile_number(self, value):
value = unidecode(value)
phone_number_validator(value)
if User.objects.filter(phone_number=value).exists():
raise Conflict(
detail=_('This mobile number is already registered as a user.'),
code='user_already_exists',
)
if InviteRequest.objects.filter(mobile_number=value).exists():
raise Conflict(
detail=_('An invite request for this mobile number already exists.'),
code='invite_request_already_exists',
)
return value
def create(self, validated_data):
invite_code = validated_data.pop('invite_code')
application = self.context['request'].auth.application
return InviteRequest.objects.create(
inviter_id=invite_code,
application=application,
**validated_data,
)
AccessToken = get_access_token_model() AccessToken = get_access_token_model()
class SessionSerializer(serializers.ModelSerializer): class SessionSerializer(serializers.ModelSerializer):
is_current = serializers.SerializerMethodField() is_current = serializers.SerializerMethodField()

121
apps/users/tests.py
View file

@ -14,7 +14,7 @@ from rest_framework.test import APIClient, APITestCase
from apps.core.models import Config from apps.core.models import Config
from apps.gooyal_oauth2.models import Scope from apps.gooyal_oauth2.models import Scope
from apps.users.models import User, InviteRequest from apps.users.models import User
AccessToken = get_access_token_model() AccessToken = get_access_token_model()
Application = get_application_model() Application = get_application_model()
@ -149,123 +149,6 @@ class UserTests(APITestCase):
self.assertEqual(response.json()['results'][0]['is_current'] , True) self.assertEqual(response.json()['results'][0]['is_current'] , True)
class InviteRequestTests(APITestCase):
user_uuid = uuid.UUID('c14e8b86-8f4a-44d9-b29d-badceb47005f')
inviter_uuid = uuid.UUID('d14e8b86-8f4a-44d9-b29d-badceb47005f')
application_uuid = uuid.UUID('e14e8b86-8f4a-44d9-b29d-badceb47005f')
access_token = 'inviteAccessToken123456789012345678901'
client_id = 'inviteClientId1234567890123456789012'
client_secret = 'inviteClientSecret123456789012345678'
mobile_number = '+989120000000'
expire_datetime = timezone.now() + timedelta(seconds=3600)
client = APIClient()
def setUp(self):
self.user = User.objects.create(pk=self.user_uuid, phone_number='+989100000001')
self.inviter = User.objects.create(pk=self.inviter_uuid, phone_number='+989100000002')
Scope.objects.create(name='accounts.invite:submit', description='accounts.invite:submit')
self.application = Application.objects.create(
name='invite-test-app',
client_id=self.client_id,
client_secret=self.client_secret,
client_type=Application.CLIENT_CONFIDENTIAL,
authorization_grant_type=Application.GRANT_CLIENT_CREDENTIALS,
hash_client_secret=False,
uuid=self.application_uuid,
user_id=self.user_uuid,
allowed_scope='accounts.invite:submit',
)
AccessToken.objects.create(
token=self.access_token,
application_id=self.application_uuid,
scope='accounts.invite:submit',
expires=self.expire_datetime,
)
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.access_token}')
def test_create_invite_request_success(self):
response = self.client.post(
reverse('users:invite_request_api'),
data={
'invite_code': str(self.inviter_uuid),
'mobile_number': self.mobile_number,
},
format='json',
)
self.assertEqual(response.status_code, 201)
self.assertEqual(InviteRequest.objects.count(), 1)
invite_request = InviteRequest.objects.get()
self.assertEqual(invite_request.inviter_id, self.inviter_uuid)
self.assertEqual(invite_request.mobile_number, self.mobile_number)
self.assertEqual(invite_request.application_id, self.application_uuid)
def test_create_invite_request_invalid_invite_code(self):
response = self.client.post(
reverse('users:invite_request_api'),
data={
'invite_code': str(uuid.uuid4()),
'mobile_number': self.mobile_number,
},
format='json',
)
self.assertEqual(response.status_code, 400)
self.assertEqual(InviteRequest.objects.count(), 0)
def test_create_invite_request_existing_user(self):
response = self.client.post(
reverse('users:invite_request_api'),
data={
'invite_code': str(self.inviter_uuid),
'mobile_number': self.inviter.phone_number,
},
format='json',
)
self.assertEqual(response.status_code, 409)
self.assertEqual(response.json()['details']['error'], 'user_already_exists')
self.assertEqual(InviteRequest.objects.count(), 0)
def test_create_invite_request_duplicate_mobile_number(self):
InviteRequest.objects.create(
inviter=self.inviter,
mobile_number=self.mobile_number,
application=self.application,
)
response = self.client.post(
reverse('users:invite_request_api'),
data={
'invite_code': str(self.inviter_uuid),
'mobile_number': self.mobile_number,
},
format='json',
)
self.assertEqual(response.status_code, 409)
self.assertEqual(response.json()['details']['error'], 'invite_request_already_exists')
self.assertEqual(InviteRequest.objects.count(), 1)
def test_create_invite_request_missing_scope(self):
self.application.allowed_scope = ''
self.application.save(update_fields=['allowed_scope'])
response = self.client.post(
reverse('users:invite_request_api'),
data={
'invite_code': str(self.inviter_uuid),
'mobile_number': self.mobile_number,
},
format='json',
)
self.assertEqual(response.status_code, 403)
self.assertEqual(InviteRequest.objects.count(), 0)

3
apps/users/urls.py
View file

@ -2,7 +2,7 @@ from django.urls import path
from django.contrib.auth.views import LogoutView from django.contrib.auth.views import LogoutView
from .views import UserListView, UserPublicRetrieveView, AccountView, RequestOTPView, ChangePasswordView, \ from .views import UserListView, UserPublicRetrieveView, AccountView, RequestOTPView, ChangePasswordView, \
OTPLoginView, ProfileDetailView, ProfileUpdateView, RequestOTTView, UserCurrentAvatarUrlView, UserInquiryView, \ OTPLoginView, ProfileDetailView, ProfileUpdateView, RequestOTTView, UserCurrentAvatarUrlView, UserInquiryView, \
UserDetailedRetrieveView, UserSessionListView, UserProfileView, InviteRequestView UserDetailedRetrieveView, UserSessionListView, UserProfileView
app_name = "users" app_name = "users"
urlpatterns = [ urlpatterns = [
@ -21,6 +21,5 @@ urlpatterns = [
path('api/request_ott/', RequestOTTView.as_view(), name='request_ott_api'), path('api/request_ott/', RequestOTTView.as_view(), name='request_ott_api'),
path('api/change_password/', ChangePasswordView.as_view(), name='change_password_api'), path('api/change_password/', ChangePasswordView.as_view(), name='change_password_api'),
path('api/inquiry/', UserInquiryView.as_view(), name='user_inquiry_api'), path('api/inquiry/', UserInquiryView.as_view(), name='user_inquiry_api'),
path('api/invite_requests/', InviteRequestView.as_view(), name='invite_request_api'),
] ]

14
apps/users/views.py
View file

@ -17,9 +17,10 @@ from rest_framework.views import APIView
from apps.gooyal_oauth2.rest_framework import IsAuthenticatedOrTokenMatchesOASRequirements from apps.gooyal_oauth2.rest_framework import IsAuthenticatedOrTokenMatchesOASRequirements
from apps.users.filters import UserFilter from apps.users.filters import UserFilter
from apps.users.forms import OTPAuthenticationForm, ProfileUpdateForm from apps.users.forms import OTPAuthenticationForm, ProfileUpdateForm
from apps.users.models import User, InviteRequest from apps.users.models import User
from apps.users.provinces_and_cities import State
from apps.users.serializers import PublicUserSerializer, AccountSerializer, RequestOTPSerializer, RequestOTTSerializer, \ from apps.users.serializers import PublicUserSerializer, AccountSerializer, RequestOTPSerializer, RequestOTTSerializer, \
ChangePasswordSerializer, UserInquirySerializer, SessionSerializer, InviteRequestSerializer ChangePasswordSerializer, UserInquirySerializer, SessionSerializer
from utils.throttles import RequestOTPDayRateThrottle, RequestOTPMinRateThrottle, NumberedRequestOTPDayRateThrottle, NumberedRequestOTPMinRateThrottle from utils.throttles import RequestOTPDayRateThrottle, RequestOTPMinRateThrottle, NumberedRequestOTPDayRateThrottle, NumberedRequestOTPMinRateThrottle
UserModel = get_user_model() UserModel = get_user_model()
@ -127,15 +128,6 @@ class UserInquiryView(generics.CreateAPIView):
required_scopes = ['accounts.profile:inquiry'] required_scopes = ['accounts.profile:inquiry']
class InviteRequestView(generics.CreateAPIView):
permission_classes = [IsAuthenticatedOrTokenMatchesOASRequirements]
required_scopes = ['accounts.invite:submit']
required_alternate_scopes ={
"POST": [["accounts.invite:submit"]],
}
serializer_class = InviteRequestSerializer
queryset = InviteRequest.objects.all()
class ChangePasswordView(generics.UpdateAPIView): class ChangePasswordView(generics.UpdateAPIView):
permission_classes = [IsAuthenticatedOrTokenHasScope] permission_classes = [IsAuthenticatedOrTokenHasScope]
serializer_class = ChangePasswordSerializer serializer_class = ChangePasswordSerializer

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

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

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

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

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

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

167
utils/clients/gooyal_promotions_client/api/api/api_v2_promotions_application_event_create.py
View file

@ -1,167 +0,0 @@
from http import HTTPStatus
from typing import Any
from urllib.parse import quote
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.event import Event
from ...types import Response
def _get_kwargs(
user_uuid: str,
*,
body: Event,
) -> dict[str, Any]:
headers: dict[str, Any] = {}
_kwargs: dict[str, Any] = {
"method": "post",
"url": "/api/v2/promotions/application/{user_uuid}/event/".format(
user_uuid=quote(str(user_uuid), safe=""),
),
}
_kwargs["json"] = body.to_dict()
headers["Content-Type"] = "application/json"
_kwargs["headers"] = headers
return _kwargs
def _parse_response(*, client: AuthenticatedClient | Client, response: httpx.Response) -> Event | None:
if response.status_code == 201:
response_201 = Event.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: AuthenticatedClient | Client, response: httpx.Response) -> Response[Event]:
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: Event,
) -> Response[Event]:
"""
Args:
user_uuid (str):
body (Event):
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[Event]
"""
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: Event,
) -> Event | None:
"""
Args:
user_uuid (str):
body (Event):
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:
Event
"""
return sync_detailed(
user_uuid=user_uuid,
client=client,
body=body,
).parsed
async def asyncio_detailed(
user_uuid: str,
*,
client: AuthenticatedClient,
body: Event,
) -> Response[Event]:
"""
Args:
user_uuid (str):
body (Event):
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[Event]
"""
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: Event,
) -> Event | None:
"""
Args:
user_uuid (str):
body (Event):
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:
Event
"""
return (
await asyncio_detailed(
user_uuid=user_uuid,
client=client,
body=body,
)
).parsed

185
utils/clients/gooyal_promotions_client/api/api/api_v2_promotions_application_plan_list.py
View file

@ -1,185 +0,0 @@
from http import HTTPStatus
from typing import Any
from urllib.parse import quote
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.paginated_user_plan_list import PaginatedUserPlanList
from ...types import UNSET, Response, Unset
def _get_kwargs(
user_uuid: str,
*,
limit: int | Unset = UNSET,
offset: int | Unset = 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": "/api/v2/promotions/application/{user_uuid}/plan/".format(
user_uuid=quote(str(user_uuid), safe=""),
),
"params": params,
}
return _kwargs
def _parse_response(*, client: AuthenticatedClient | Client, response: httpx.Response) -> PaginatedUserPlanList | None:
if response.status_code == 200:
response_200 = PaginatedUserPlanList.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: AuthenticatedClient | Client, response: httpx.Response
) -> Response[PaginatedUserPlanList]:
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: int | Unset = UNSET,
offset: int | Unset = UNSET,
) -> Response[PaginatedUserPlanList]:
"""
Args:
user_uuid (str):
limit (int | Unset):
offset (int | Unset):
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[PaginatedUserPlanList]
"""
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: int | Unset = UNSET,
offset: int | Unset = UNSET,
) -> PaginatedUserPlanList | None:
"""
Args:
user_uuid (str):
limit (int | Unset):
offset (int | Unset):
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:
PaginatedUserPlanList
"""
return sync_detailed(
user_uuid=user_uuid,
client=client,
limit=limit,
offset=offset,
).parsed
async def asyncio_detailed(
user_uuid: str,
*,
client: AuthenticatedClient,
limit: int | Unset = UNSET,
offset: int | Unset = UNSET,
) -> Response[PaginatedUserPlanList]:
"""
Args:
user_uuid (str):
limit (int | Unset):
offset (int | Unset):
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[PaginatedUserPlanList]
"""
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: int | Unset = UNSET,
offset: int | Unset = UNSET,
) -> PaginatedUserPlanList | None:
"""
Args:
user_uuid (str):
limit (int | Unset):
offset (int | Unset):
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:
PaginatedUserPlanList
"""
return (
await asyncio_detailed(
user_uuid=user_uuid,
client=client,
limit=limit,
offset=offset,
)
).parsed

181
utils/clients/gooyal_promotions_client/api/api/api_v2_promotions_application_plan_promote_create.py
View file

@ -1,181 +0,0 @@
from http import HTTPStatus
from typing import Any
from urllib.parse import quote
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.promote import Promote
from ...types import Response
def _get_kwargs(
user_uuid: str,
id: str,
*,
body: Promote,
) -> dict[str, Any]:
headers: dict[str, Any] = {}
_kwargs: dict[str, Any] = {
"method": "post",
"url": "/api/v2/promotions/application/{user_uuid}/plan/{id}/promote/".format(
user_uuid=quote(str(user_uuid), safe=""),
id=quote(str(id), safe=""),
),
}
_kwargs["json"] = body.to_dict()
headers["Content-Type"] = "application/json"
_kwargs["headers"] = headers
return _kwargs
def _parse_response(*, client: AuthenticatedClient | Client, response: httpx.Response) -> Promote | None:
if response.status_code == 200:
response_200 = Promote.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: AuthenticatedClient | Client, response: httpx.Response) -> Response[Promote]:
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,
id: str,
*,
client: AuthenticatedClient,
body: Promote,
) -> Response[Promote]:
"""
Args:
user_uuid (str):
id (str):
body (Promote):
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[Promote]
"""
kwargs = _get_kwargs(
user_uuid=user_uuid,
id=id,
body=body,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
user_uuid: str,
id: str,
*,
client: AuthenticatedClient,
body: Promote,
) -> Promote | None:
"""
Args:
user_uuid (str):
id (str):
body (Promote):
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:
Promote
"""
return sync_detailed(
user_uuid=user_uuid,
id=id,
client=client,
body=body,
).parsed
async def asyncio_detailed(
user_uuid: str,
id: str,
*,
client: AuthenticatedClient,
body: Promote,
) -> Response[Promote]:
"""
Args:
user_uuid (str):
id (str):
body (Promote):
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[Promote]
"""
kwargs = _get_kwargs(
user_uuid=user_uuid,
id=id,
body=body,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
user_uuid: str,
id: str,
*,
client: AuthenticatedClient,
body: Promote,
) -> Promote | None:
"""
Args:
user_uuid (str):
id (str):
body (Promote):
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:
Promote
"""
return (
await asyncio_detailed(
user_uuid=user_uuid,
id=id,
client=client,
body=body,
)
).parsed

161
utils/clients/gooyal_promotions_client/api/api/api_v2_promotions_application_plan_retrieve.py
View file

@ -1,161 +0,0 @@
from http import HTTPStatus
from typing import Any
from urllib.parse import quote
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.user_plan import UserPlan
from ...types import Response
def _get_kwargs(
user_uuid: str,
id: str,
) -> dict[str, Any]:
_kwargs: dict[str, Any] = {
"method": "get",
"url": "/api/v2/promotions/application/{user_uuid}/plan/{id}/".format(
user_uuid=quote(str(user_uuid), safe=""),
id=quote(str(id), safe=""),
),
}
return _kwargs
def _parse_response(*, client: AuthenticatedClient | Client, response: httpx.Response) -> UserPlan | None:
if response.status_code == 200:
response_200 = UserPlan.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: AuthenticatedClient | Client, response: httpx.Response) -> Response[UserPlan]:
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,
id: str,
*,
client: AuthenticatedClient,
) -> Response[UserPlan]:
"""
Args:
user_uuid (str):
id (str):
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[UserPlan]
"""
kwargs = _get_kwargs(
user_uuid=user_uuid,
id=id,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
user_uuid: str,
id: str,
*,
client: AuthenticatedClient,
) -> UserPlan | None:
"""
Args:
user_uuid (str):
id (str):
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:
UserPlan
"""
return sync_detailed(
user_uuid=user_uuid,
id=id,
client=client,
).parsed
async def asyncio_detailed(
user_uuid: str,
id: str,
*,
client: AuthenticatedClient,
) -> Response[UserPlan]:
"""
Args:
user_uuid (str):
id (str):
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[UserPlan]
"""
kwargs = _get_kwargs(
user_uuid=user_uuid,
id=id,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
user_uuid: str,
id: str,
*,
client: AuthenticatedClient,
) -> UserPlan | None:
"""
Args:
user_uuid (str):
id (str):
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:
UserPlan
"""
return (
await asyncio_detailed(
user_uuid=user_uuid,
id=id,
client=client,
)
).parsed

1
utils/clients/gooyal_promotions_client/api/promotions/__init__.py
View file

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

151
utils/clients/gooyal_promotions_client/api/promotions/promotions_api_v1_events_submit_create.py
View file

@ -1,151 +0,0 @@
from http import HTTPStatus
from typing import Any
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.promote import Promote
from ...types import Response
def _get_kwargs(
*,
body: Promote,
) -> dict[str, Any]:
headers: dict[str, Any] = {}
_kwargs: dict[str, Any] = {
"method": "post",
"url": "/promotions/api/v1/events/submit",
}
_kwargs["json"] = body.to_dict()
headers["Content-Type"] = "application/json"
_kwargs["headers"] = headers
return _kwargs
def _parse_response(*, client: AuthenticatedClient | Client, response: httpx.Response) -> Promote | None:
if response.status_code == 201:
response_201 = Promote.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: AuthenticatedClient | Client, response: httpx.Response) -> Response[Promote]:
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,
body: Promote,
) -> Response[Promote]:
"""
Args:
body (Promote):
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[Promote]
"""
kwargs = _get_kwargs(
body=body,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
*,
client: AuthenticatedClient,
body: Promote,
) -> Promote | None:
"""
Args:
body (Promote):
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:
Promote
"""
return sync_detailed(
client=client,
body=body,
).parsed
async def asyncio_detailed(
*,
client: AuthenticatedClient,
body: Promote,
) -> Response[Promote]:
"""
Args:
body (Promote):
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[Promote]
"""
kwargs = _get_kwargs(
body=body,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
*,
client: AuthenticatedClient,
body: Promote,
) -> Promote | None:
"""
Args:
body (Promote):
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:
Promote
"""
return (
await asyncio_detailed(
client=client,
body=body,
)
).parsed

167
utils/clients/gooyal_promotions_client/api/promotions/promotions_api_v1_plans_create.py
View file

@ -1,167 +0,0 @@
from http import HTTPStatus
from typing import Any
from urllib.parse import quote
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.promote import Promote
from ...types import Response
def _get_kwargs(
plan: str,
*,
body: Promote,
) -> dict[str, Any]:
headers: dict[str, Any] = {}
_kwargs: dict[str, Any] = {
"method": "post",
"url": "/promotions/api/v1/plans/{plan}/".format(
plan=quote(str(plan), safe=""),
),
}
_kwargs["json"] = body.to_dict()
headers["Content-Type"] = "application/json"
_kwargs["headers"] = headers
return _kwargs
def _parse_response(*, client: AuthenticatedClient | Client, response: httpx.Response) -> Promote | None:
if response.status_code == 201:
response_201 = Promote.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: AuthenticatedClient | Client, response: httpx.Response) -> Response[Promote]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)
def sync_detailed(
plan: str,
*,
client: AuthenticatedClient,
body: Promote,
) -> Response[Promote]:
"""
Args:
plan (str):
body (Promote):
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[Promote]
"""
kwargs = _get_kwargs(
plan=plan,
body=body,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
plan: str,
*,
client: AuthenticatedClient,
body: Promote,
) -> Promote | None:
"""
Args:
plan (str):
body (Promote):
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:
Promote
"""
return sync_detailed(
plan=plan,
client=client,
body=body,
).parsed
async def asyncio_detailed(
plan: str,
*,
client: AuthenticatedClient,
body: Promote,
) -> Response[Promote]:
"""
Args:
plan (str):
body (Promote):
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[Promote]
"""
kwargs = _get_kwargs(
plan=plan,
body=body,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
plan: str,
*,
client: AuthenticatedClient,
body: Promote,
) -> Promote | None:
"""
Args:
plan (str):
body (Promote):
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:
Promote
"""
return (
await asyncio_detailed(
plan=plan,
client=client,
body=body,
)
).parsed

185
utils/clients/gooyal_promotions_client/api/promotions/promotions_api_v1_plans_promotins_list.py
View file

@ -1,185 +0,0 @@
from http import HTTPStatus
from typing import Any
from urllib.parse import quote
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.paginated_promotion_list import PaginatedPromotionList
from ...types import UNSET, Response, Unset
def _get_kwargs(
plan: str,
*,
limit: int | Unset = UNSET,
offset: int | Unset = 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": "/promotions/api/v1/plans/{plan}/promotins".format(
plan=quote(str(plan), safe=""),
),
"params": params,
}
return _kwargs
def _parse_response(*, client: AuthenticatedClient | Client, response: httpx.Response) -> PaginatedPromotionList | None:
if response.status_code == 200:
response_200 = PaginatedPromotionList.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: AuthenticatedClient | Client, response: httpx.Response
) -> Response[PaginatedPromotionList]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)
def sync_detailed(
plan: str,
*,
client: AuthenticatedClient,
limit: int | Unset = UNSET,
offset: int | Unset = UNSET,
) -> Response[PaginatedPromotionList]:
"""
Args:
plan (str):
limit (int | Unset):
offset (int | Unset):
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[PaginatedPromotionList]
"""
kwargs = _get_kwargs(
plan=plan,
limit=limit,
offset=offset,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
plan: str,
*,
client: AuthenticatedClient,
limit: int | Unset = UNSET,
offset: int | Unset = UNSET,
) -> PaginatedPromotionList | None:
"""
Args:
plan (str):
limit (int | Unset):
offset (int | Unset):
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:
PaginatedPromotionList
"""
return sync_detailed(
plan=plan,
client=client,
limit=limit,
offset=offset,
).parsed
async def asyncio_detailed(
plan: str,
*,
client: AuthenticatedClient,
limit: int | Unset = UNSET,
offset: int | Unset = UNSET,
) -> Response[PaginatedPromotionList]:
"""
Args:
plan (str):
limit (int | Unset):
offset (int | Unset):
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[PaginatedPromotionList]
"""
kwargs = _get_kwargs(
plan=plan,
limit=limit,
offset=offset,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
plan: str,
*,
client: AuthenticatedClient,
limit: int | Unset = UNSET,
offset: int | Unset = UNSET,
) -> PaginatedPromotionList | None:
"""
Args:
plan (str):
limit (int | Unset):
offset (int | Unset):
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:
PaginatedPromotionList
"""
return (
await asyncio_detailed(
plan=plan,
client=client,
limit=limit,
offset=offset,
)
).parsed

167
utils/clients/gooyal_promotions_client/api/promotions/promotions_api_v2_plans_promote_create.py
View file

@ -1,167 +0,0 @@
from http import HTTPStatus
from typing import Any
from urllib.parse import quote
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.promote import Promote
from ...types import Response
def _get_kwargs(
id: str,
*,
body: Promote,
) -> dict[str, Any]:
headers: dict[str, Any] = {}
_kwargs: dict[str, Any] = {
"method": "post",
"url": "/promotions/api/v2/plans/{id}/promote/".format(
id=quote(str(id), safe=""),
),
}
_kwargs["json"] = body.to_dict()
headers["Content-Type"] = "application/json"
_kwargs["headers"] = headers
return _kwargs
def _parse_response(*, client: AuthenticatedClient | Client, response: httpx.Response) -> Promote | None:
if response.status_code == 200:
response_200 = Promote.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: AuthenticatedClient | Client, response: httpx.Response) -> Response[Promote]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)
def sync_detailed(
id: str,
*,
client: AuthenticatedClient,
body: Promote,
) -> Response[Promote]:
"""
Args:
id (str):
body (Promote):
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[Promote]
"""
kwargs = _get_kwargs(
id=id,
body=body,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
id: str,
*,
client: AuthenticatedClient,
body: Promote,
) -> Promote | None:
"""
Args:
id (str):
body (Promote):
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:
Promote
"""
return sync_detailed(
id=id,
client=client,
body=body,
).parsed
async def asyncio_detailed(
id: str,
*,
client: AuthenticatedClient,
body: Promote,
) -> Response[Promote]:
"""
Args:
id (str):
body (Promote):
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[Promote]
"""
kwargs = _get_kwargs(
id=id,
body=body,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
id: str,
*,
client: AuthenticatedClient,
body: Promote,
) -> Promote | None:
"""
Args:
id (str):
body (Promote):
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:
Promote
"""
return (
await asyncio_detailed(
id=id,
client=client,
body=body,
)
).parsed

147
utils/clients/gooyal_promotions_client/api/promotions/promotions_api_v2_plans_retrieve.py
View file

@ -1,147 +0,0 @@
from http import HTTPStatus
from typing import Any
from urllib.parse import quote
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.user_plan import UserPlan
from ...types import Response
def _get_kwargs(
id: str,
) -> dict[str, Any]:
_kwargs: dict[str, Any] = {
"method": "get",
"url": "/promotions/api/v2/plans/{id}/".format(
id=quote(str(id), safe=""),
),
}
return _kwargs
def _parse_response(*, client: AuthenticatedClient | Client, response: httpx.Response) -> UserPlan | None:
if response.status_code == 200:
response_200 = UserPlan.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: AuthenticatedClient | Client, response: httpx.Response) -> Response[UserPlan]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)
def sync_detailed(
id: str,
*,
client: AuthenticatedClient,
) -> Response[UserPlan]:
"""
Args:
id (str):
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[UserPlan]
"""
kwargs = _get_kwargs(
id=id,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
id: str,
*,
client: AuthenticatedClient,
) -> UserPlan | None:
"""
Args:
id (str):
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:
UserPlan
"""
return sync_detailed(
id=id,
client=client,
).parsed
async def asyncio_detailed(
id: str,
*,
client: AuthenticatedClient,
) -> Response[UserPlan]:
"""
Args:
id (str):
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[UserPlan]
"""
kwargs = _get_kwargs(
id=id,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
id: str,
*,
client: AuthenticatedClient,
) -> UserPlan | None:
"""
Args:
id (str):
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:
UserPlan
"""
return (
await asyncio_detailed(
id=id,
client=client,
)
).parsed

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

@ -1,268 +0,0 @@
import ssl
from typing import Any
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: httpx.Timeout | None = field(default=None, kw_only=True, alias="timeout")
_verify_ssl: 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: httpx.Client | None = field(default=None, init=False)
_async_client: httpx.AsyncClient | None = 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 configuration"""
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 set 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: httpx.Timeout | None = field(default=None, kw_only=True, alias="timeout")
_verify_ssl: 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: httpx.Client | None = field(default=None, init=False)
_async_client: httpx.AsyncClient | None = 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 configuration"""
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 set 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_promotions_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"]

21
utils/clients/gooyal_promotions_client/models/__init__.py
View file

@ -1,21 +0,0 @@
"""Contains all the data models used in inputs/outputs"""
from .event import Event
from .paginated_promotion_list import PaginatedPromotionList
from .paginated_user_plan_list import PaginatedUserPlanList
from .promote import Promote
from .promotion import Promotion
from .state_enum import StateEnum
from .user_plan import UserPlan
from .user_recipient import UserRecipient
__all__ = (
"Event",
"PaginatedPromotionList",
"PaginatedUserPlanList",
"Promote",
"Promotion",
"StateEnum",
"UserPlan",
"UserRecipient",
)

120
utils/clients/gooyal_promotions_client/models/event.py
View file

@ -1,120 +0,0 @@
from __future__ import annotations
from collections.abc import Mapping
from typing import Any, TypeVar, cast
from uuid import UUID
from attrs import define as _attrs_define
from attrs import field as _attrs_field
T = TypeVar("T", bound="Event")
@_attrs_define
class Event:
"""
Attributes:
label (str):
data (Any):
user (None | UUID):
application (None | UUID):
"""
label: str
data: Any
user: None | UUID
application: None | UUID
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
def to_dict(self) -> dict[str, Any]:
label = self.label
data = self.data
user: None | str
if isinstance(self.user, UUID):
user = str(self.user)
else:
user = self.user
application: None | str
if isinstance(self.application, UUID):
application = str(self.application)
else:
application = self.application
field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"label": label,
"data": data,
"user": user,
"application": application,
}
)
return field_dict
@classmethod
def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
d = dict(src_dict)
label = d.pop("label")
data = d.pop("data")
def _parse_user(data: object) -> None | UUID:
if data is None:
return data
try:
if not isinstance(data, str):
raise TypeError()
user_type_0 = UUID(data)
return user_type_0
except (TypeError, ValueError, AttributeError, KeyError):
pass
return cast(None | UUID, data)
user = _parse_user(d.pop("user"))
def _parse_application(data: object) -> 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 (TypeError, ValueError, AttributeError, KeyError):
pass
return cast(None | UUID, data)
application = _parse_application(d.pop("application"))
event = cls(
label=label,
data=data,
user=user,
application=application,
)
event.additional_properties = d
return event
@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

125
utils/clients/gooyal_promotions_client/models/paginated_promotion_list.py
View file

@ -1,125 +0,0 @@
from __future__ import annotations
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, TypeVar, 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.promotion import Promotion
T = TypeVar("T", bound="PaginatedPromotionList")
@_attrs_define
class PaginatedPromotionList:
"""
Attributes:
count (int): Example: 123.
results (list[Promotion]):
next_ (None | str | Unset): Example: http://api.example.org/accounts/?offset=400&limit=100.
previous (None | str | Unset): Example: http://api.example.org/accounts/?offset=200&limit=100.
"""
count: int
results: list[Promotion]
next_: None | str | Unset = UNSET
previous: None | str | Unset = 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_: None | str | Unset
if isinstance(self.next_, Unset):
next_ = UNSET
else:
next_ = self.next_
previous: None | str | Unset
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.promotion import Promotion
d = dict(src_dict)
count = d.pop("count")
results = []
_results = d.pop("results")
for results_item_data in _results:
results_item = Promotion.from_dict(results_item_data)
results.append(results_item)
def _parse_next_(data: object) -> None | str | Unset:
if data is None:
return data
if isinstance(data, Unset):
return data
return cast(None | str | Unset, data)
next_ = _parse_next_(d.pop("next", UNSET))
def _parse_previous(data: object) -> None | str | Unset:
if data is None:
return data
if isinstance(data, Unset):
return data
return cast(None | str | Unset, data)
previous = _parse_previous(d.pop("previous", UNSET))
paginated_promotion_list = cls(
count=count,
results=results,
next_=next_,
previous=previous,
)
paginated_promotion_list.additional_properties = d
return paginated_promotion_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

125
utils/clients/gooyal_promotions_client/models/paginated_user_plan_list.py
View file

@ -1,125 +0,0 @@
from __future__ import annotations
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, TypeVar, 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.user_plan import UserPlan
T = TypeVar("T", bound="PaginatedUserPlanList")
@_attrs_define
class PaginatedUserPlanList:
"""
Attributes:
count (int): Example: 123.
results (list[UserPlan]):
next_ (None | str | Unset): Example: http://api.example.org/accounts/?offset=400&limit=100.
previous (None | str | Unset): Example: http://api.example.org/accounts/?offset=200&limit=100.
"""
count: int
results: list[UserPlan]
next_: None | str | Unset = UNSET
previous: None | str | Unset = 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_: None | str | Unset
if isinstance(self.next_, Unset):
next_ = UNSET
else:
next_ = self.next_
previous: None | str | Unset
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.user_plan import UserPlan
d = dict(src_dict)
count = d.pop("count")
results = []
_results = d.pop("results")
for results_item_data in _results:
results_item = UserPlan.from_dict(results_item_data)
results.append(results_item)
def _parse_next_(data: object) -> None | str | Unset:
if data is None:
return data
if isinstance(data, Unset):
return data
return cast(None | str | Unset, data)
next_ = _parse_next_(d.pop("next", UNSET))
def _parse_previous(data: object) -> None | str | Unset:
if data is None:
return data
if isinstance(data, Unset):
return data
return cast(None | str | Unset, data)
previous = _parse_previous(d.pop("previous", UNSET))
paginated_user_plan_list = cls(
count=count,
results=results,
next_=next_,
previous=previous,
)
paginated_user_plan_list.additional_properties = d
return paginated_user_plan_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

91
utils/clients/gooyal_promotions_client/models/promote.py
View file

@ -1,91 +0,0 @@
from __future__ import annotations
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, TypeVar
from attrs import define as _attrs_define
from attrs import field as _attrs_field
if TYPE_CHECKING:
from ..models.promotion import Promotion
T = TypeVar("T", bound="Promote")
@_attrs_define
class Promote:
"""
Attributes:
promotions (list[Promotion]):
label (str):
data (Any):
"""
promotions: list[Promotion]
label: str
data: Any
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
def to_dict(self) -> dict[str, Any]:
promotions = []
for promotions_item_data in self.promotions:
promotions_item = promotions_item_data.to_dict()
promotions.append(promotions_item)
label = self.label
data = self.data
field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"promotions": promotions,
"label": label,
"data": data,
}
)
return field_dict
@classmethod
def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
from ..models.promotion import Promotion
d = dict(src_dict)
promotions = []
_promotions = d.pop("promotions")
for promotions_item_data in _promotions:
promotions_item = Promotion.from_dict(promotions_item_data)
promotions.append(promotions_item)
label = d.pop("label")
data = d.pop("data")
promote = cls(
promotions=promotions,
label=label,
data=data,
)
promote.additional_properties = d
return promote
@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

129
utils/clients/gooyal_promotions_client/models/promotion.py
View file

@ -1,129 +0,0 @@
from __future__ import annotations
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, TypeVar, cast
from uuid import UUID
from attrs import define as _attrs_define
from attrs import field as _attrs_field
from ..models.state_enum import StateEnum
from ..types import UNSET, Unset
if TYPE_CHECKING:
from ..models.event import Event
T = TypeVar("T", bound="Promotion")
@_attrs_define
class Promotion:
"""
Attributes:
uuid (UUID):
event (Event):
user (UUID):
promotion_amount (int | None | Unset):
state (StateEnum | Unset): * `1` - created
* `2` - delayed
* `3` - pending
* `4` - incomplete
* `5` - success
* `6` - failed
* `7` - expected_failure
"""
uuid: UUID
event: Event
user: UUID
promotion_amount: int | None | Unset = UNSET
state: StateEnum | Unset = UNSET
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
def to_dict(self) -> dict[str, Any]:
uuid = str(self.uuid)
event = self.event.to_dict()
user = str(self.user)
promotion_amount: int | None | Unset
if isinstance(self.promotion_amount, Unset):
promotion_amount = UNSET
else:
promotion_amount = self.promotion_amount
state: int | Unset = UNSET
if not isinstance(self.state, Unset):
state = self.state.value
field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"uuid": uuid,
"event": event,
"user": user,
}
)
if promotion_amount is not UNSET:
field_dict["promotion_amount"] = promotion_amount
if state is not UNSET:
field_dict["state"] = state
return field_dict
@classmethod
def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
from ..models.event import Event
d = dict(src_dict)
uuid = UUID(d.pop("uuid"))
event = Event.from_dict(d.pop("event"))
user = UUID(d.pop("user"))
def _parse_promotion_amount(data: object) -> int | None | Unset:
if data is None:
return data
if isinstance(data, Unset):
return data
return cast(int | None | Unset, data)
promotion_amount = _parse_promotion_amount(d.pop("promotion_amount", UNSET))
_state = d.pop("state", UNSET)
state: StateEnum | Unset
if isinstance(_state, Unset):
state = UNSET
else:
state = StateEnum(_state)
promotion = cls(
uuid=uuid,
event=event,
user=user,
promotion_amount=promotion_amount,
state=state,
)
promotion.additional_properties = d
return promotion
@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

14
utils/clients/gooyal_promotions_client/models/state_enum.py
View file

@ -1,14 +0,0 @@
from enum import IntEnum
class StateEnum(IntEnum):
VALUE_1 = 1
VALUE_2 = 2
VALUE_3 = 3
VALUE_4 = 4
VALUE_5 = 5
VALUE_6 = 6
VALUE_7 = 7
def __str__(self) -> str:
return str(self.value)

113
utils/clients/gooyal_promotions_client/models/user_plan.py
View file

@ -1,113 +0,0 @@
from __future__ import annotations
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, TypeVar, cast
from attrs import define as _attrs_define
from attrs import field as _attrs_field
if TYPE_CHECKING:
from ..models.user_recipient import UserRecipient
T = TypeVar("T", bound="UserPlan")
@_attrs_define
class UserPlan:
"""
Attributes:
title (str):
banner (None | str):
description (str):
description_details (Any):
recipients (list[UserRecipient]):
"""
title: str
banner: None | str
description: str
description_details: Any
recipients: list[UserRecipient]
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
def to_dict(self) -> dict[str, Any]:
title = self.title
banner: None | str
banner = self.banner
description = self.description
description_details = self.description_details
recipients = []
for recipients_item_data in self.recipients:
recipients_item = recipients_item_data.to_dict()
recipients.append(recipients_item)
field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"title": title,
"banner": banner,
"description": description,
"description_details": description_details,
"recipients": recipients,
}
)
return field_dict
@classmethod
def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
from ..models.user_recipient import UserRecipient
d = dict(src_dict)
title = d.pop("title")
def _parse_banner(data: object) -> None | str:
if data is None:
return data
return cast(None | str, data)
banner = _parse_banner(d.pop("banner"))
description = d.pop("description")
description_details = d.pop("description_details")
recipients = []
_recipients = d.pop("recipients")
for recipients_item_data in _recipients:
recipients_item = UserRecipient.from_dict(recipients_item_data)
recipients.append(recipients_item)
user_plan = cls(
title=title,
banner=banner,
description=description,
description_details=description_details,
recipients=recipients,
)
user_plan.additional_properties = d
return user_plan
@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

69
utils/clients/gooyal_promotions_client/models/user_recipient.py
View file

@ -1,69 +0,0 @@
from __future__ import annotations
from collections.abc import Mapping
from typing import Any, TypeVar
from attrs import define as _attrs_define
from attrs import field as _attrs_field
T = TypeVar("T", bound="UserRecipient")
@_attrs_define
class UserRecipient:
"""
Attributes:
label (str):
base_amount_field (str):
"""
label: str
base_amount_field: str
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
def to_dict(self) -> dict[str, Any]:
label = self.label
base_amount_field = self.base_amount_field
field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"label": label,
"base_amount_field": base_amount_field,
}
)
return field_dict
@classmethod
def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
d = dict(src_dict)
label = d.pop("label")
base_amount_field = d.pop("base_amount_field")
user_recipient = cls(
label=label,
base_amount_field=base_amount_field,
)
user_recipient.additional_properties = d
return user_recipient
@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_promotions_client/py.typed
View file

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

54
utils/clients/gooyal_promotions_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, TypeVar
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 = IO[bytes] | bytes | str
FileTypes = (
# (filename, file (or bytes), content_type)
tuple[str | None, FileContent, str | None]
# (filename, file (or bytes), content_type, headers)
| tuple[str | None, FileContent, str | None, Mapping[str, str]]
)
RequestFiles = list[tuple[str, FileTypes]]
@define
class File:
"""Contains information for file uploads"""
payload: BinaryIO
file_name: str | None = None
mime_type: str | None = 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: T | None
__all__ = ["UNSET", "File", "FileTypes", "RequestFiles", "Response", "Unset"]