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

basic promote api

Browse source
This commit is contained in:
Sayyid Hamid Mahdavi 2025-08-03 15:54:22 +03:30
parent 75924dd2b6
commit 18a866ca02
40 changed files with 3464 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

58
apps/promotions/handlers.py
View file

@ -1,7 +1,12 @@
import logging
from django.db import models
from django.utils.translation import gettext_lazy as _
from enum import property as enum_property # ??
from utils.clients.wallet_client import deposit_to_user_wallet_submit, deposit_to_user_wallet_verify
logger = logging.getLogger("promotion_handler")
class PromotionTypeChoices(models.TextChoices):
WALLET_PERCENTAGE = 'wallet percentage', _('Wallet percentage')
@ -20,24 +25,69 @@ class PromotionTypeChoices(models.TextChoices):
class BasePromotionHandler:
def __init__(self, plan):
self.plan = plan
def __init__(self, application=None, policy=None, balance=None):
self.application = application
self.policy = policy
self.balance = balance
def calculate(self, **kwargs):
raise NotImplementedError()
def promote(self, **kwargs):
raise NotImplementedError()
class WalletPercentageHandler(BasePromotionHandler):
payee_type = 2
payee_uuid = ''
def calculate(self, **kwargs):
base_amount = kwargs.get('base_amount', 0)
if base_amount <= 1000 and self.plan.balance >= 1000:
if base_amount <= 1000 and self.balance >= 1000:
promotion_amount = base_amount
else:
promotion_amount = 1000 if self.plan.balance >= 1000 else self.plan.balance
promotion_amount = 1000 if self.balance >= 1000 else self.balance
return promotion_amount
def promote(self, promotion, **kwargs):
return True
# promotion_amount = self.calculate(**kwargs)
try:
data = {
"uuid": str(promotion.uuid),
"payee_id": str(self.payee_uuid),
"payee_type": self.payee_type,
"amount": promotion.promotion_amount,
"details": {
'description': "for promotion",
'reference_id': str(promotion.uuid),
'application_details_url': ''
},
}
submit_response = deposit_to_user_wallet_submit(data)
try:
submit_response.uuid
except Exception as e:
logger.exception('Failed to submit payment')
raise Exception('Failed to submit payment')
verify_response = deposit_to_user_wallet_verify(str(promotion.uuid))
logger.debug(f'verify response: {verify_response}')
try:
if not verify_response.uuid or verify_response.state != 5:
logger.debug('Failed to verify payment')
raise Exception('Failed to verify payment')
else:
return True
except Exception as e:
logger.exception('Failed to verify payment')
raise Exception('Failed to verify payment')
except Exception as e:
logger.exception('error in finish visit:')
raise e
class OtherHandler(BasePromotionHandler):
pass

23
apps/promotions/migrations/0002_promotion_promotion_amount_alter_promotion_state.py Normal file
View file

@ -0,0 +1,23 @@
# Generated by Django 5.2.4 on 2025-08-03 12:02
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('promotions', '0001_initial'),
]
operations = [
migrations.AddField(
model_name='promotion',
name='promotion_amount',
field=models.IntegerField(blank=True, null=True),
),
migrations.AlterField(
model_name='promotion',
name='state',
field=models.IntegerField(choices=[(0, 'Created'), (1, 'Assigned'), (2, 'Promoted'), (10, 'Expired')], default=0),
),
]

18
apps/promotions/migrations/0003_alter_plan_balance.py Normal file
View file

@ -0,0 +1,18 @@
# Generated by Django 5.2.4 on 2025-08-03 12:03
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('promotions', '0002_promotion_promotion_amount_alter_promotion_state'),
]
operations = [
migrations.AlterField(
model_name='plan',
name='balance',
field=models.IntegerField(default=0, verbose_name='balance'),
),
]

18
apps/promotions/migrations/0004_promotion_base_amount.py Normal file
View file

@ -0,0 +1,18 @@
# Generated by Django 5.2.4 on 2025-08-03 12:19
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('promotions', '0003_alter_plan_balance'),
]
operations = [
migrations.AddField(
model_name='promotion',
name='base_amount',
field=models.IntegerField(blank=True, null=True),
),
]

56
apps/promotions/models.py
View file

@ -1,5 +1,10 @@
from cgitb import handler
from django.conf import settings
from django.db import models
from django.db import models, transaction
from django.db.models import F
from django.utils import timezone
from oauth2_provider.settings import oauth2_settings
from apps.promotions.handlers import PromotionTypeChoices, BasePromotionHandler
@ -23,7 +28,7 @@ class Plan(BaseModel):
description = models.TextField(verbose_name=_('description'), blank=True)
promotion_type = models.CharField(max_length=64, verbose_name=_('promotion type'), db_index=True, choices=PromotionTypeChoices.choices)
policy = models.JSONField(default=dict, verbose_name=_('policy'), null=True, blank=True)
balance = models.DecimalField(max_digits=10, decimal_places=2, verbose_name=_('balance'), default=0)
balance = models.IntegerField(verbose_name=_('balance'), default=0)
objects = PlanQuerySet.as_manager()
@ -44,6 +49,14 @@ class Plan(BaseModel):
self.promotion_amount = handler.calculate(**kwargs)
return self.promotion_amount
def reserve_promotion(self, promotion):
updated = False
if self.balance >= promotion.promotion_amount:
updated = Plan.objects.filter(pk=self.pk, balance__gte=promotion.promotion_amount).update(
balance=F('balance') - promotion.promotion_amount
)
return updated
class PromotionQuerySet(models.QuerySet):
def used(self, user):
@ -54,6 +67,7 @@ class Promotion(BaseModel):
class StateChoices(models.IntegerChoices):
CREATED = 0, _('Created')
ASSIGNED = 1, _('Assigned')
PROMOTED = 2, _('Promoted')
EXPIRED = 10, _('Expired')
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT)
@ -61,8 +75,44 @@ class Promotion(BaseModel):
related_name='+', null=True, blank=True)
plan = models.ForeignKey(Plan, on_delete=models.PROTECT, related_name='promotions', null=True, blank=True)
state = models.IntegerField(choices=StateChoices.choices, default=StateChoices.CREATED)
base_amount = models.IntegerField(null=True, blank=True)
promotion_amount = models.IntegerField(null=True, blank=True)
data = models.JSONField(null=True, blank=True)
objects = PromotionQuerySet.as_manager()
def change_state(self, from_states: list | tuple, to_state, same_ok=True, raise_exception=False,
error_message=None,
data: None | dict = None):
data = data or {}
data['updated_at'] = timezone.now()
if same_ok and self.state == to_state:
return self
if self.state in from_states:
updated = Promotion.objects.filter(pk=self.pk, state__in=from_states).update(
state=to_state,
**data
)
if updated: # double check in db level
self.state = to_state
return self
if raise_exception:
raise Exception(error_message)
def promote(self, **kwargs):
with transaction.atomic():
self.plan.reserve_promotion(self)
self.change_state(from_states=[Promotion.StateChoices.CREATED], to_state=Promotion.StateChoices.ASSIGNED,
same_ok=False, raise_exception=True)
handler = self.plan.handler
if handler.promote(promotion=self, **kwargs):
self.change_state(from_states=[Promotion.StateChoices.ASSIGNED], to_state=Promotion.StateChoices.PROMOTED)
return True
return False

22
apps/promotions/serializers.py
View file

@ -21,17 +21,24 @@ class PlanSerializer(serializers.ModelSerializer):
read_only_fields = ('application', "promotion_amount")
class PromotionSerializer(serializers.ModelSerializer):
base_amount = serializers.IntegerField(required=True)
promotion_amount = serializers.IntegerField(read_only=True)
class Meta:
model = Promotion
fields = (
"base_amount",
"promotion_amount",
"plan",
"user",
"application",
)
read_only_fields = ('user','application')
read_only_fields = ('user','application', "promotion_amount",)
class CalculatedPlanSerializer(serializers.ModelSerializer):
base_amount = serializers.IntegerField(required=True)
promotion_amount = serializers.IntegerField(read_only=True)
class Meta:
model = Promotion
fields = (
@ -44,4 +51,15 @@ class CalculatedPlanSerializer(serializers.ModelSerializer):
# def get_promotion_amount(self, obj):
# user = self.context['request'].user
# return obj.calculate_promotion(user)
# return obj.calculate_promotion(user)
class PlanPromotSerializer(serializers.ModelSerializer):
class Meta:
model = Promotion
fields = (
"user",
"application",
)
read_only_fields = ('user','application')

4
apps/promotions/urls.py
View file

@ -8,10 +8,10 @@ from django.urls import NoReverseMatch, path, re_path, include
app_name = 'promotions'
application_router = DefaultRouter()
application_router.register(r'', views.ApplicationPlanViewSet, basename='application' )
application_router.register(r'promotion', views.ApplicationPromotionViewSet, basename='application' )
user_router = DefaultRouter()
user_router.register('', views.UserPlanViewSet, basename='user')
user_router.register('plan', views.UserPlanViewSet, basename='user')
urlpatterns = [
path('api/v1/application/<user_uuid>/', include(application_router.urls), name='application'),

64
apps/promotions/views.py
View file

@ -10,13 +10,12 @@ from rest_framework.generics import get_object_or_404
from rest_framework.viewsets import GenericViewSet
from .models import Promotion, Plan
from .serializers import PlanSerializer, PromotionSerializer, CalculatedPlanSerializer
from .serializers import PlanSerializer, PromotionSerializer, CalculatedPlanSerializer, PlanPromotSerializer
from apps.gooyal_oauth2.models import Application
from apps.gooyal_oauth2.rest_framework import IsAuthenticatedOrTokenMatchesOASRequirements
from apps.gooyal_oauth2.utils import get_application
from ..users.models import User
from django.shortcuts import render
from django.http import Http404
@ -34,29 +33,64 @@ from apps.gooyal_oauth2.utils import get_application
from ..users.models import User
class ApplicationPlanViewSet(mixins.ListModelMixin,
mixins.CreateModelMixin,
GenericViewSet):
queryset = Plan.objects.all()
serializer_class = PlanSerializer
# class ApplicationPlanViewSet(mixins.ListModelMixin,
# mixins.CreateModelMixin,
# GenericViewSet):
# queryset = Plan.objects.all()
# serializer_class = PlanSerializer
#
# permission_classes = [IsAuthenticatedOrTokenMatchesOASRequirements]
# required_alternate_scopes = {
# "POST": [["promotions.application.user-plans:submit"]],
# "GET": [["promotions.application.user-promotions:list-retrieve"]],
# }
#
# def get_queryset(self):
# user_uuid = self.kwargs.get('user_uuid')
# user = User.objects.get(uuid=user_uuid)
# application = get_application(self.request)
# return Plan.objects.filter(application=application)
#
class ApplicationPromotionViewSet(
mixins.RetrieveModelMixin,
mixins.ListModelMixin,
mixins.CreateModelMixin,
GenericViewSet
):
serializer_class = PromotionSerializer
permission_classes = [IsAuthenticatedOrTokenMatchesOASRequirements]
required_alternate_scopes = {
"POST": [["promotions.application.user-plans:submit"]],
"POST": [["promotions.application.user-promotions:promote"]],
"GET": [["promotions.application.user-promotions:list-retrieve"]],
}
# def perform_create(self, serializer):
# user_uuid = self.kwargs.get('user_uuid')
# user = User.objects.get(uuid=user_uuid)
# application = get_application(self.request)
# instance = serializer.save(user=user, application=application)
def get_queryset(self):
user_uuid = self.kwargs.get('user_uuid')
user = User.objects.get(uuid=user_uuid)
application = get_application(self.request)
return Promotion.objects.filter(application=application, user=user)
def perform_create(self, serializer):
user_uuid = self.kwargs.get('user_uuid')
user = User.objects.get(uuid=user_uuid)
application = get_application(self.request)
plan = serializer.validated_data['plan']
promotion_args = dict(user=user, application=application, **serializer.validated_data)
promotion_amount = plan.calculate_promotion(**promotion_args)
if promotion_amount:
promotion = serializer.save(promotion_amount=promotion_amount, user=user, application=application)
promotion.promote(**promotion_args)
class UserPlanViewSet(mixins.RetrieveModelMixin,
mixins.ListModelMixin,
GenericViewSet):
GenericViewSet):
serializer_class = PlanSerializer
permission_classes = [IsAuthenticatedOrTokenMatchesOASRequirements]
required_alternate_scopes = {

8
gooyal_wallet_client/__init__.py Normal file
View file

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

1
gooyal_wallet_client/api/__init__.py Normal file
View file

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

0
gooyal_wallet_client/api/wallet/__init__.py Normal file
View file

126
gooyal_wallet_client/api/wallet/wallet_api_application_balance_retrieve.py Normal file
View file

@ -0,0 +1,126 @@
from http import HTTPStatus
from typing import Any, Optional, Union
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.application_balance import ApplicationBalance
from ...types import Response
def _get_kwargs() -> dict[str, Any]:
_kwargs: dict[str, Any] = {
"method": "get",
"url": "/wallet/api/application/balance",
}
return _kwargs
def _parse_response(
*, client: Union[AuthenticatedClient, Client], response: httpx.Response
) -> Optional[ApplicationBalance]:
if response.status_code == 200:
response_200 = ApplicationBalance.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[ApplicationBalance]:
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[ApplicationBalance]:
"""API for getting specified user or application balance.
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[ApplicationBalance]
"""
kwargs = _get_kwargs()
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
*,
client: AuthenticatedClient,
) -> Optional[ApplicationBalance]:
"""API for getting specified user or application balance.
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:
ApplicationBalance
"""
return sync_detailed(
client=client,
).parsed
async def asyncio_detailed(
*,
client: AuthenticatedClient,
) -> Response[ApplicationBalance]:
"""API for getting specified user or application balance.
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[ApplicationBalance]
"""
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[ApplicationBalance]:
"""API for getting specified user or application balance.
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:
ApplicationBalance
"""
return (
await asyncio_detailed(
client=client,
)
).parsed

155
gooyal_wallet_client/api/wallet/wallet_api_application_deposit_create.py Normal file
View file

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

146
gooyal_wallet_client/api/wallet/wallet_api_application_deposit_verify_retrieve.py Normal file
View file

@ -0,0 +1,146 @@
from http import HTTPStatus
from typing import Any, Optional, Union
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.application_deposit import ApplicationDeposit
from ...types import Response
def _get_kwargs(
uuid: str,
) -> dict[str, Any]:
_kwargs: dict[str, Any] = {
"method": "get",
"url": f"/wallet/api/application/deposit/{uuid}/verify",
}
return _kwargs
def _parse_response(
*, client: Union[AuthenticatedClient, Client], response: httpx.Response
) -> Optional[ApplicationDeposit]:
if response.status_code == 200:
response_200 = ApplicationDeposit.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[ApplicationDeposit]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)
def sync_detailed(
uuid: str,
*,
client: AuthenticatedClient,
) -> Response[ApplicationDeposit]:
"""
Args:
uuid (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[ApplicationDeposit]
"""
kwargs = _get_kwargs(
uuid=uuid,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
uuid: str,
*,
client: AuthenticatedClient,
) -> Optional[ApplicationDeposit]:
"""
Args:
uuid (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:
ApplicationDeposit
"""
return sync_detailed(
uuid=uuid,
client=client,
).parsed
async def asyncio_detailed(
uuid: str,
*,
client: AuthenticatedClient,
) -> Response[ApplicationDeposit]:
"""
Args:
uuid (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[ApplicationDeposit]
"""
kwargs = _get_kwargs(
uuid=uuid,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
uuid: str,
*,
client: AuthenticatedClient,
) -> Optional[ApplicationDeposit]:
"""
Args:
uuid (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:
ApplicationDeposit
"""
return (
await asyncio_detailed(
uuid=uuid,
client=client,
)
).parsed

154
gooyal_wallet_client/api/wallet/wallet_api_application_list.py Normal file
View file

@ -0,0 +1,154 @@
from http import HTTPStatus
from typing import Any, Optional, Union
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.paginated_application_transaction_list import PaginatedApplicationTransactionList
from ...types import UNSET, Response, Unset
def _get_kwargs(
*,
page: Union[Unset, int] = UNSET,
) -> dict[str, Any]:
params: dict[str, Any] = {}
params["page"] = page
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": "/wallet/api/application/",
"params": params,
}
return _kwargs
def _parse_response(
*, client: Union[AuthenticatedClient, Client], response: httpx.Response
) -> Optional[PaginatedApplicationTransactionList]:
if response.status_code == 200:
response_200 = PaginatedApplicationTransactionList.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[PaginatedApplicationTransactionList]:
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,
page: Union[Unset, int] = UNSET,
) -> Response[PaginatedApplicationTransactionList]:
"""
Args:
page (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[PaginatedApplicationTransactionList]
"""
kwargs = _get_kwargs(
page=page,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
*,
client: AuthenticatedClient,
page: Union[Unset, int] = UNSET,
) -> Optional[PaginatedApplicationTransactionList]:
"""
Args:
page (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:
PaginatedApplicationTransactionList
"""
return sync_detailed(
client=client,
page=page,
).parsed
async def asyncio_detailed(
*,
client: AuthenticatedClient,
page: Union[Unset, int] = UNSET,
) -> Response[PaginatedApplicationTransactionList]:
"""
Args:
page (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[PaginatedApplicationTransactionList]
"""
kwargs = _get_kwargs(
page=page,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
*,
client: AuthenticatedClient,
page: Union[Unset, int] = UNSET,
) -> Optional[PaginatedApplicationTransactionList]:
"""
Args:
page (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:
PaginatedApplicationTransactionList
"""
return (
await asyncio_detailed(
client=client,
page=page,
)
).parsed

146
gooyal_wallet_client/api/wallet/wallet_api_application_retrieve.py Normal file
View file

@ -0,0 +1,146 @@
from http import HTTPStatus
from typing import Any, Optional, Union
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.application_transaction import ApplicationTransaction
from ...types import Response
def _get_kwargs(
uuid: str,
) -> dict[str, Any]:
_kwargs: dict[str, Any] = {
"method": "get",
"url": f"/wallet/api/application/{uuid}/",
}
return _kwargs
def _parse_response(
*, client: Union[AuthenticatedClient, Client], response: httpx.Response
) -> Optional[ApplicationTransaction]:
if response.status_code == 200:
response_200 = ApplicationTransaction.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[ApplicationTransaction]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)
def sync_detailed(
uuid: str,
*,
client: AuthenticatedClient,
) -> Response[ApplicationTransaction]:
"""
Args:
uuid (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[ApplicationTransaction]
"""
kwargs = _get_kwargs(
uuid=uuid,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
uuid: str,
*,
client: AuthenticatedClient,
) -> Optional[ApplicationTransaction]:
"""
Args:
uuid (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:
ApplicationTransaction
"""
return sync_detailed(
uuid=uuid,
client=client,
).parsed
async def asyncio_detailed(
uuid: str,
*,
client: AuthenticatedClient,
) -> Response[ApplicationTransaction]:
"""
Args:
uuid (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[ApplicationTransaction]
"""
kwargs = _get_kwargs(
uuid=uuid,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
uuid: str,
*,
client: AuthenticatedClient,
) -> Optional[ApplicationTransaction]:
"""
Args:
uuid (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:
ApplicationTransaction
"""
return (
await asyncio_detailed(
uuid=uuid,
client=client,
)
).parsed

155
gooyal_wallet_client/api/wallet/wallet_api_application_withdraw_create.py Normal file
View file

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

146
gooyal_wallet_client/api/wallet/wallet_api_application_withdraw_verify_retrieve.py Normal file
View file

@ -0,0 +1,146 @@
from http import HTTPStatus
from typing import Any, Optional, Union
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.application_withdraw import ApplicationWithdraw
from ...types import Response
def _get_kwargs(
uuid: str,
) -> dict[str, Any]:
_kwargs: dict[str, Any] = {
"method": "get",
"url": f"/wallet/api/application/withdraw/{uuid}/verify",
}
return _kwargs
def _parse_response(
*, client: Union[AuthenticatedClient, Client], response: httpx.Response
) -> Optional[ApplicationWithdraw]:
if response.status_code == 200:
response_200 = ApplicationWithdraw.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[ApplicationWithdraw]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)
def sync_detailed(
uuid: str,
*,
client: AuthenticatedClient,
) -> Response[ApplicationWithdraw]:
"""
Args:
uuid (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[ApplicationWithdraw]
"""
kwargs = _get_kwargs(
uuid=uuid,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
uuid: str,
*,
client: AuthenticatedClient,
) -> Optional[ApplicationWithdraw]:
"""
Args:
uuid (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:
ApplicationWithdraw
"""
return sync_detailed(
uuid=uuid,
client=client,
).parsed
async def asyncio_detailed(
uuid: str,
*,
client: AuthenticatedClient,
) -> Response[ApplicationWithdraw]:
"""
Args:
uuid (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[ApplicationWithdraw]
"""
kwargs = _get_kwargs(
uuid=uuid,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
uuid: str,
*,
client: AuthenticatedClient,
) -> Optional[ApplicationWithdraw]:
"""
Args:
uuid (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:
ApplicationWithdraw
"""
return (
await asyncio_detailed(
uuid=uuid,
client=client,
)
).parsed

174
gooyal_wallet_client/api/wallet/wallet_api_transactions_list.py Normal file
View file

@ -0,0 +1,174 @@
from http import HTTPStatus
from typing import Any, Optional, Union
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.paginated_user_transaction_list import PaginatedUserTransactionList
from ...types import UNSET, Response, Unset
def _get_kwargs(
*,
page: Union[Unset, int] = UNSET,
) -> dict[str, Any]:
params: dict[str, Any] = {}
params["page"] = page
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": "/wallet/api/transactions/",
"params": params,
}
return _kwargs
def _parse_response(
*, client: Union[AuthenticatedClient, Client], response: httpx.Response
) -> Optional[PaginatedUserTransactionList]:
if response.status_code == 200:
response_200 = PaginatedUserTransactionList.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[PaginatedUserTransactionList]:
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,
page: Union[Unset, int] = UNSET,
) -> Response[PaginatedUserTransactionList]:
"""GET: API for getting list of transactions for authenticated user
POST: API for creating transaction for transferring funds from authenticated user to application
owner. (deprecated)
args:
- user: authenticated user
Args:
page (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[PaginatedUserTransactionList]
"""
kwargs = _get_kwargs(
page=page,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
*,
client: AuthenticatedClient,
page: Union[Unset, int] = UNSET,
) -> Optional[PaginatedUserTransactionList]:
"""GET: API for getting list of transactions for authenticated user
POST: API for creating transaction for transferring funds from authenticated user to application
owner. (deprecated)
args:
- user: authenticated user
Args:
page (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:
PaginatedUserTransactionList
"""
return sync_detailed(
client=client,
page=page,
).parsed
async def asyncio_detailed(
*,
client: AuthenticatedClient,
page: Union[Unset, int] = UNSET,
) -> Response[PaginatedUserTransactionList]:
"""GET: API for getting list of transactions for authenticated user
POST: API for creating transaction for transferring funds from authenticated user to application
owner. (deprecated)
args:
- user: authenticated user
Args:
page (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[PaginatedUserTransactionList]
"""
kwargs = _get_kwargs(
page=page,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
*,
client: AuthenticatedClient,
page: Union[Unset, int] = UNSET,
) -> Optional[PaginatedUserTransactionList]:
"""GET: API for getting list of transactions for authenticated user
POST: API for creating transaction for transferring funds from authenticated user to application
owner. (deprecated)
args:
- user: authenticated user
Args:
page (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:
PaginatedUserTransactionList
"""
return (
await asyncio_detailed(
client=client,
page=page,
)
).parsed

122
gooyal_wallet_client/api/wallet/wallet_api_wallet_balance_retrieve.py Normal file
View file

@ -0,0 +1,122 @@
from http import HTTPStatus
from typing import Any, Optional, Union
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.my_wallet import MyWallet
from ...types import Response
def _get_kwargs() -> dict[str, Any]:
_kwargs: dict[str, Any] = {
"method": "get",
"url": "/wallet/api/wallet/balance",
}
return _kwargs
def _parse_response(*, client: Union[AuthenticatedClient, Client], response: httpx.Response) -> Optional[MyWallet]:
if response.status_code == 200:
response_200 = MyWallet.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[MyWallet]:
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[MyWallet]:
"""API for getting authenticated user balance.
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[MyWallet]
"""
kwargs = _get_kwargs()
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
*,
client: AuthenticatedClient,
) -> Optional[MyWallet]:
"""API for getting authenticated user balance.
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:
MyWallet
"""
return sync_detailed(
client=client,
).parsed
async def asyncio_detailed(
*,
client: AuthenticatedClient,
) -> Response[MyWallet]:
"""API for getting authenticated user balance.
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[MyWallet]
"""
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[MyWallet]:
"""API for getting authenticated user balance.
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:
MyWallet
"""
return (
await asyncio_detailed(
client=client,
)
).parsed

268
gooyal_wallet_client/client.py Normal file
View file

@ -0,0 +1,268 @@
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
gooyal_wallet_client/errors.py Normal file
View file

@ -0,0 +1,16 @@
"""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"]

27
gooyal_wallet_client/models/__init__.py Normal file
View file

@ -0,0 +1,27 @@
"""Contains all the data models used in inputs/outputs"""
from .application_balance import ApplicationBalance
from .application_deposit import ApplicationDeposit
from .application_transaction import ApplicationTransaction
from .application_withdraw import ApplicationWithdraw
from .my_wallet import MyWallet
from .paginated_application_transaction_list import PaginatedApplicationTransactionList
from .paginated_user_transaction_list import PaginatedUserTransactionList
from .payee_type_enum import PayeeTypeEnum
from .payer_type_enum import PayerTypeEnum
from .state_enum import StateEnum
from .user_transaction import UserTransaction
__all__ = (
"ApplicationBalance",
"ApplicationDeposit",
"ApplicationTransaction",
"ApplicationWithdraw",
"MyWallet",
"PaginatedApplicationTransactionList",
"PaginatedUserTransactionList",
"PayeeTypeEnum",
"PayerTypeEnum",
"StateEnum",
"UserTransaction",
)

58
gooyal_wallet_client/models/application_balance.py Normal file
View file

@ -0,0 +1,58 @@
from typing import Any, TypeVar, Union
from attrs import define as _attrs_define
from attrs import field as _attrs_field
from ..types import UNSET, Unset
T = TypeVar("T", bound="ApplicationBalance")
@_attrs_define
class ApplicationBalance:
"""
Attributes:
balance (Union[Unset, int]):
"""
balance: Union[Unset, int] = UNSET
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
def to_dict(self) -> dict[str, Any]:
balance = self.balance
field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update({})
if balance is not UNSET:
field_dict["balance"] = balance
return field_dict
@classmethod
def from_dict(cls: type[T], src_dict: dict[str, Any]) -> T:
d = src_dict.copy()
balance = d.pop("balance", UNSET)
application_balance = cls(
balance=balance,
)
application_balance.additional_properties = d
return application_balance
@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

235
gooyal_wallet_client/models/application_deposit.py Normal file
View file

@ -0,0 +1,235 @@
import datetime
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 dateutil.parser import isoparse
from ..models.payee_type_enum import PayeeTypeEnum
from ..models.payer_type_enum import PayerTypeEnum
from ..models.state_enum import StateEnum
from ..types import UNSET, Unset
T = TypeVar("T", bound="ApplicationDeposit")
@_attrs_define
class ApplicationDeposit:
"""
Attributes:
uuid (UUID):
payer_id (Union[None, UUID]):
payer_type (Union[None, PayerTypeEnum]):
payee_id (UUID):
created_at (Union[None, datetime.datetime]):
state (StateEnum): * `1` - created
* `2` - delayed
* `3` - pending
* `4` - incomplete
* `5` - success
* `6` - failed
* `7` - expected_failure
amount (int):
updated_at (Union[None, datetime.datetime]):
payee_type (Union[None, PayeeTypeEnum, Unset]):
details (Union[Unset, Any]):
"""
uuid: UUID
payer_id: Union[None, UUID]
payer_type: Union[None, PayerTypeEnum]
payee_id: UUID
created_at: Union[None, datetime.datetime]
state: StateEnum
amount: int
updated_at: Union[None, datetime.datetime]
payee_type: Union[None, PayeeTypeEnum, Unset] = UNSET
details: Union[Unset, Any] = UNSET
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
def to_dict(self) -> dict[str, Any]:
uuid = str(self.uuid)
payer_id: Union[None, str]
if isinstance(self.payer_id, UUID):
payer_id = str(self.payer_id)
else:
payer_id = self.payer_id
payer_type: Union[None, int]
if isinstance(self.payer_type, PayerTypeEnum):
payer_type = self.payer_type.value
else:
payer_type = self.payer_type
payee_id = str(self.payee_id)
created_at: Union[None, str]
if isinstance(self.created_at, datetime.datetime):
created_at = self.created_at.isoformat()
else:
created_at = self.created_at
state = self.state.value
amount = self.amount
updated_at: Union[None, str]
if isinstance(self.updated_at, datetime.datetime):
updated_at = self.updated_at.isoformat()
else:
updated_at = self.updated_at
payee_type: Union[None, Unset, int]
if isinstance(self.payee_type, Unset):
payee_type = UNSET
elif isinstance(self.payee_type, PayeeTypeEnum):
payee_type = self.payee_type.value
else:
payee_type = self.payee_type
details = self.details
field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"uuid": uuid,
"payer_id": payer_id,
"payer_type": payer_type,
"payee_id": payee_id,
"created_at": created_at,
"state": state,
"amount": amount,
"updated_at": updated_at,
}
)
if payee_type is not UNSET:
field_dict["payee_type"] = payee_type
if details is not UNSET:
field_dict["details"] = details
return field_dict
@classmethod
def from_dict(cls: type[T], src_dict: dict[str, Any]) -> T:
d = src_dict.copy()
uuid = UUID(d.pop("uuid"))
def _parse_payer_id(data: object) -> Union[None, UUID]:
if data is None:
return data
try:
if not isinstance(data, str):
raise TypeError()
payer_id_type_0 = UUID(data)
return payer_id_type_0
except: # noqa: E722
pass
return cast(Union[None, UUID], data)
payer_id = _parse_payer_id(d.pop("payer_id"))
def _parse_payer_type(data: object) -> Union[None, PayerTypeEnum]:
if data is None:
return data
try:
if not isinstance(data, int):
raise TypeError()
payer_type_type_0 = PayerTypeEnum(data)
return payer_type_type_0
except: # noqa: E722
pass
return cast(Union[None, PayerTypeEnum], data)
payer_type = _parse_payer_type(d.pop("payer_type"))
payee_id = UUID(d.pop("payee_id"))
def _parse_created_at(data: object) -> Union[None, datetime.datetime]:
if data is None:
return data
try:
if not isinstance(data, str):
raise TypeError()
created_at_type_0 = isoparse(data)
return created_at_type_0
except: # noqa: E722
pass
return cast(Union[None, datetime.datetime], data)
created_at = _parse_created_at(d.pop("created_at"))
state = StateEnum(d.pop("state"))
amount = d.pop("amount")
def _parse_updated_at(data: object) -> Union[None, datetime.datetime]:
if data is None:
return data
try:
if not isinstance(data, str):
raise TypeError()
updated_at_type_0 = isoparse(data)
return updated_at_type_0
except: # noqa: E722
pass
return cast(Union[None, datetime.datetime], data)
updated_at = _parse_updated_at(d.pop("updated_at"))
def _parse_payee_type(data: object) -> Union[None, PayeeTypeEnum, Unset]:
if data is None:
return data
if isinstance(data, Unset):
return data
try:
if not isinstance(data, int):
raise TypeError()
payee_type_type_0 = PayeeTypeEnum(data)
return payee_type_type_0
except: # noqa: E722
pass
return cast(Union[None, PayeeTypeEnum, Unset], data)
payee_type = _parse_payee_type(d.pop("payee_type", UNSET))
details = d.pop("details", UNSET)
application_deposit = cls(
uuid=uuid,
payer_id=payer_id,
payer_type=payer_type,
payee_id=payee_id,
created_at=created_at,
state=state,
amount=amount,
updated_at=updated_at,
payee_type=payee_type,
details=details,
)
application_deposit.additional_properties = d
return application_deposit
@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

238
gooyal_wallet_client/models/application_transaction.py Normal file
View file

@ -0,0 +1,238 @@
import datetime
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 dateutil.parser import isoparse
from ..models.payer_type_enum import PayerTypeEnum
from ..models.state_enum import StateEnum
from ..types import UNSET, Unset
T = TypeVar("T", bound="ApplicationTransaction")
@_attrs_define
class ApplicationTransaction:
"""
Attributes:
uuid (UUID):
payee_id (UUID):
payee_type (UUID):
created_at (Union[None, datetime.datetime]):
state (StateEnum): * `1` - created
* `2` - delayed
* `3` - pending
* `4` - incomplete
* `5` - success
* `6` - failed
* `7` - expected_failure
amount (int):
updated_at (Union[None, datetime.datetime]):
details (Any):
role (str):
payer_id (Union[None, UUID, Unset]):
payer_type (Union[None, PayerTypeEnum, Unset]):
delay (Union[Unset, int]):
"""
uuid: UUID
payee_id: UUID
payee_type: UUID
created_at: Union[None, datetime.datetime]
state: StateEnum
amount: int
updated_at: Union[None, datetime.datetime]
details: Any
role: str
payer_id: Union[None, UUID, Unset] = UNSET
payer_type: Union[None, PayerTypeEnum, Unset] = UNSET
delay: Union[Unset, int] = UNSET
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
def to_dict(self) -> dict[str, Any]:
uuid = str(self.uuid)
payee_id = str(self.payee_id)
payee_type = str(self.payee_type)
created_at: Union[None, str]
if isinstance(self.created_at, datetime.datetime):
created_at = self.created_at.isoformat()
else:
created_at = self.created_at
state = self.state.value
amount = self.amount
updated_at: Union[None, str]
if isinstance(self.updated_at, datetime.datetime):
updated_at = self.updated_at.isoformat()
else:
updated_at = self.updated_at
details = self.details
role = self.role
payer_id: Union[None, Unset, str]
if isinstance(self.payer_id, Unset):
payer_id = UNSET
elif isinstance(self.payer_id, UUID):
payer_id = str(self.payer_id)
else:
payer_id = self.payer_id
payer_type: Union[None, Unset, int]
if isinstance(self.payer_type, Unset):
payer_type = UNSET
elif isinstance(self.payer_type, PayerTypeEnum):
payer_type = self.payer_type.value
else:
payer_type = self.payer_type
delay = self.delay
field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"uuid": uuid,
"payee_id": payee_id,
"payee_type": payee_type,
"created_at": created_at,
"state": state,
"amount": amount,
"updated_at": updated_at,
"details": details,
"role": role,
}
)
if payer_id is not UNSET:
field_dict["payer_id"] = payer_id
if payer_type is not UNSET:
field_dict["payer_type"] = payer_type
if delay is not UNSET:
field_dict["delay"] = delay
return field_dict
@classmethod
def from_dict(cls: type[T], src_dict: dict[str, Any]) -> T:
d = src_dict.copy()
uuid = UUID(d.pop("uuid"))
payee_id = UUID(d.pop("payee_id"))
payee_type = UUID(d.pop("payee_type"))
def _parse_created_at(data: object) -> Union[None, datetime.datetime]:
if data is None:
return data
try:
if not isinstance(data, str):
raise TypeError()
created_at_type_0 = isoparse(data)
return created_at_type_0
except: # noqa: E722
pass
return cast(Union[None, datetime.datetime], data)
created_at = _parse_created_at(d.pop("created_at"))
state = StateEnum(d.pop("state"))
amount = d.pop("amount")
def _parse_updated_at(data: object) -> Union[None, datetime.datetime]:
if data is None:
return data
try:
if not isinstance(data, str):
raise TypeError()
updated_at_type_0 = isoparse(data)
return updated_at_type_0
except: # noqa: E722
pass
return cast(Union[None, datetime.datetime], data)
updated_at = _parse_updated_at(d.pop("updated_at"))
details = d.pop("details")
role = d.pop("role")
def _parse_payer_id(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()
payer_id_type_0 = UUID(data)
return payer_id_type_0
except: # noqa: E722
pass
return cast(Union[None, UUID, Unset], data)
payer_id = _parse_payer_id(d.pop("payer_id", UNSET))
def _parse_payer_type(data: object) -> Union[None, PayerTypeEnum, Unset]:
if data is None:
return data
if isinstance(data, Unset):
return data
try:
if not isinstance(data, int):
raise TypeError()
payer_type_type_0 = PayerTypeEnum(data)
return payer_type_type_0
except: # noqa: E722
pass
return cast(Union[None, PayerTypeEnum, Unset], data)
payer_type = _parse_payer_type(d.pop("payer_type", UNSET))
delay = d.pop("delay", UNSET)
application_transaction = cls(
uuid=uuid,
payee_id=payee_id,
payee_type=payee_type,
created_at=created_at,
state=state,
amount=amount,
updated_at=updated_at,
details=details,
role=role,
payer_id=payer_id,
payer_type=payer_type,
delay=delay,
)
application_transaction.additional_properties = d
return application_transaction
@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

239
gooyal_wallet_client/models/application_withdraw.py Normal file
View file

@ -0,0 +1,239 @@
import datetime
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 dateutil.parser import isoparse
from ..models.payee_type_enum import PayeeTypeEnum
from ..models.payer_type_enum import PayerTypeEnum
from ..models.state_enum import StateEnum
from ..types import UNSET, Unset
T = TypeVar("T", bound="ApplicationWithdraw")
@_attrs_define
class ApplicationWithdraw:
"""
Attributes:
uuid (UUID):
payee_id (UUID):
payee_type (Union[None, PayeeTypeEnum]):
created_at (Union[None, datetime.datetime]):
state (StateEnum): * `1` - created
* `2` - delayed
* `3` - pending
* `4` - incomplete
* `5` - success
* `6` - failed
* `7` - expected_failure
amount (int):
updated_at (Union[None, datetime.datetime]):
details (Any):
payer_id (Union[None, UUID, Unset]):
payer_type (Union[None, PayerTypeEnum, Unset]):
"""
uuid: UUID
payee_id: UUID
payee_type: Union[None, PayeeTypeEnum]
created_at: Union[None, datetime.datetime]
state: StateEnum
amount: int
updated_at: Union[None, datetime.datetime]
details: Any
payer_id: Union[None, UUID, Unset] = UNSET
payer_type: Union[None, PayerTypeEnum, Unset] = UNSET
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
def to_dict(self) -> dict[str, Any]:
uuid = str(self.uuid)
payee_id = str(self.payee_id)
payee_type: Union[None, int]
if isinstance(self.payee_type, PayeeTypeEnum):
payee_type = self.payee_type.value
else:
payee_type = self.payee_type
created_at: Union[None, str]
if isinstance(self.created_at, datetime.datetime):
created_at = self.created_at.isoformat()
else:
created_at = self.created_at
state = self.state.value
amount = self.amount
updated_at: Union[None, str]
if isinstance(self.updated_at, datetime.datetime):
updated_at = self.updated_at.isoformat()
else:
updated_at = self.updated_at
details = self.details
payer_id: Union[None, Unset, str]
if isinstance(self.payer_id, Unset):
payer_id = UNSET
elif isinstance(self.payer_id, UUID):
payer_id = str(self.payer_id)
else:
payer_id = self.payer_id
payer_type: Union[None, Unset, int]
if isinstance(self.payer_type, Unset):
payer_type = UNSET
elif isinstance(self.payer_type, PayerTypeEnum):
payer_type = self.payer_type.value
else:
payer_type = self.payer_type
field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"uuid": uuid,
"payee_id": payee_id,
"payee_type": payee_type,
"created_at": created_at,
"state": state,
"amount": amount,
"updated_at": updated_at,
"details": details,
}
)
if payer_id is not UNSET:
field_dict["payer_id"] = payer_id
if payer_type is not UNSET:
field_dict["payer_type"] = payer_type
return field_dict
@classmethod
def from_dict(cls: type[T], src_dict: dict[str, Any]) -> T:
d = src_dict.copy()
uuid = UUID(d.pop("uuid"))
payee_id = UUID(d.pop("payee_id"))
def _parse_payee_type(data: object) -> Union[None, PayeeTypeEnum]:
if data is None:
return data
try:
if not isinstance(data, int):
raise TypeError()
payee_type_type_0 = PayeeTypeEnum(data)
return payee_type_type_0
except: # noqa: E722
pass
return cast(Union[None, PayeeTypeEnum], data)
payee_type = _parse_payee_type(d.pop("payee_type"))
def _parse_created_at(data: object) -> Union[None, datetime.datetime]:
if data is None:
return data
try:
if not isinstance(data, str):
raise TypeError()
created_at_type_0 = isoparse(data)
return created_at_type_0
except: # noqa: E722
pass
return cast(Union[None, datetime.datetime], data)
created_at = _parse_created_at(d.pop("created_at"))
state = StateEnum(d.pop("state"))
amount = d.pop("amount")
def _parse_updated_at(data: object) -> Union[None, datetime.datetime]:
if data is None:
return data
try:
if not isinstance(data, str):
raise TypeError()
updated_at_type_0 = isoparse(data)
return updated_at_type_0
except: # noqa: E722
pass
return cast(Union[None, datetime.datetime], data)
updated_at = _parse_updated_at(d.pop("updated_at"))
details = d.pop("details")
def _parse_payer_id(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()
payer_id_type_0 = UUID(data)
return payer_id_type_0
except: # noqa: E722
pass
return cast(Union[None, UUID, Unset], data)
payer_id = _parse_payer_id(d.pop("payer_id", UNSET))
def _parse_payer_type(data: object) -> Union[None, PayerTypeEnum, Unset]:
if data is None:
return data
if isinstance(data, Unset):
return data
try:
if not isinstance(data, int):
raise TypeError()
payer_type_type_0 = PayerTypeEnum(data)
return payer_type_type_0
except: # noqa: E722
pass
return cast(Union[None, PayerTypeEnum, Unset], data)
payer_type = _parse_payer_type(d.pop("payer_type", UNSET))
application_withdraw = cls(
uuid=uuid,
payee_id=payee_id,
payee_type=payee_type,
created_at=created_at,
state=state,
amount=amount,
updated_at=updated_at,
details=details,
payer_id=payer_id,
payer_type=payer_type,
)
application_withdraw.additional_properties = d
return application_withdraw
@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

58
gooyal_wallet_client/models/my_wallet.py Normal file
View file

@ -0,0 +1,58 @@
from typing import Any, TypeVar, Union
from attrs import define as _attrs_define
from attrs import field as _attrs_field
from ..types import UNSET, Unset
T = TypeVar("T", bound="MyWallet")
@_attrs_define
class MyWallet:
"""
Attributes:
balance (Union[Unset, int]):
"""
balance: Union[Unset, int] = UNSET
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
def to_dict(self) -> dict[str, Any]:
balance = self.balance
field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update({})
if balance is not UNSET:
field_dict["balance"] = balance
return field_dict
@classmethod
def from_dict(cls: type[T], src_dict: dict[str, Any]) -> T:
d = src_dict.copy()
balance = d.pop("balance", UNSET)
my_wallet = cls(
balance=balance,
)
my_wallet.additional_properties = d
return my_wallet
@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

122
gooyal_wallet_client/models/paginated_application_transaction_list.py Normal file
View file

@ -0,0 +1,122 @@
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.application_transaction import ApplicationTransaction
T = TypeVar("T", bound="PaginatedApplicationTransactionList")
@_attrs_define
class PaginatedApplicationTransactionList:
"""
Attributes:
count (int): Example: 123.
results (list['ApplicationTransaction']):
next_ (Union[None, Unset, str]): Example: http://api.example.org/accounts/?page=4.
previous (Union[None, Unset, str]): Example: http://api.example.org/accounts/?page=2.
"""
count: int
results: list["ApplicationTransaction"]
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: dict[str, Any]) -> T:
from ..models.application_transaction import ApplicationTransaction
d = src_dict.copy()
count = d.pop("count")
results = []
_results = d.pop("results")
for results_item_data in _results:
results_item = ApplicationTransaction.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_application_transaction_list = cls(
count=count,
results=results,
next_=next_,
previous=previous,
)
paginated_application_transaction_list.additional_properties = d
return paginated_application_transaction_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

122
gooyal_wallet_client/models/paginated_user_transaction_list.py Normal file
View file

@ -0,0 +1,122 @@
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.user_transaction import UserTransaction
T = TypeVar("T", bound="PaginatedUserTransactionList")
@_attrs_define
class PaginatedUserTransactionList:
"""
Attributes:
count (int): Example: 123.
results (list['UserTransaction']):
next_ (Union[None, Unset, str]): Example: http://api.example.org/accounts/?page=4.
previous (Union[None, Unset, str]): Example: http://api.example.org/accounts/?page=2.
"""
count: int
results: list["UserTransaction"]
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: dict[str, Any]) -> T:
from ..models.user_transaction import UserTransaction
d = src_dict.copy()
count = d.pop("count")
results = []
_results = d.pop("results")
for results_item_data in _results:
results_item = UserTransaction.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_user_transaction_list = cls(
count=count,
results=results,
next_=next_,
previous=previous,
)
paginated_user_transaction_list.additional_properties = d
return paginated_user_transaction_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

9
gooyal_wallet_client/models/payee_type_enum.py Normal file
View file

@ -0,0 +1,9 @@
from enum import IntEnum
class PayeeTypeEnum(IntEnum):
VALUE_1 = 1
VALUE_2 = 2
def __str__(self) -> str:
return str(self.value)

9
gooyal_wallet_client/models/payer_type_enum.py Normal file
View file

@ -0,0 +1,9 @@
from enum import IntEnum
class PayerTypeEnum(IntEnum):
VALUE_1 = 1
VALUE_2 = 2
def __str__(self) -> str:
return str(self.value)

14
gooyal_wallet_client/models/state_enum.py Normal file
View file

@ -0,0 +1,14 @@
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)

238
gooyal_wallet_client/models/user_transaction.py Normal file
View file

@ -0,0 +1,238 @@
import datetime
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 dateutil.parser import isoparse
from ..models.payer_type_enum import PayerTypeEnum
from ..models.state_enum import StateEnum
from ..types import UNSET, Unset
T = TypeVar("T", bound="UserTransaction")
@_attrs_define
class UserTransaction:
"""
Attributes:
uuid (UUID):
payee_id (UUID):
payee_type (UUID):
created_at (Union[None, datetime.datetime]):
state (StateEnum): * `1` - created
* `2` - delayed
* `3` - pending
* `4` - incomplete
* `5` - success
* `6` - failed
* `7` - expected_failure
amount (int):
updated_at (Union[None, datetime.datetime]):
details (Any):
role (int):
payer_id (Union[None, UUID, Unset]):
payer_type (Union[None, PayerTypeEnum, Unset]):
delay (Union[Unset, int]):
"""
uuid: UUID
payee_id: UUID
payee_type: UUID
created_at: Union[None, datetime.datetime]
state: StateEnum
amount: int
updated_at: Union[None, datetime.datetime]
details: Any
role: int
payer_id: Union[None, UUID, Unset] = UNSET
payer_type: Union[None, PayerTypeEnum, Unset] = UNSET
delay: Union[Unset, int] = UNSET
additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
def to_dict(self) -> dict[str, Any]:
uuid = str(self.uuid)
payee_id = str(self.payee_id)
payee_type = str(self.payee_type)
created_at: Union[None, str]
if isinstance(self.created_at, datetime.datetime):
created_at = self.created_at.isoformat()
else:
created_at = self.created_at
state = self.state.value
amount = self.amount
updated_at: Union[None, str]
if isinstance(self.updated_at, datetime.datetime):
updated_at = self.updated_at.isoformat()
else:
updated_at = self.updated_at
details = self.details
role = self.role
payer_id: Union[None, Unset, str]
if isinstance(self.payer_id, Unset):
payer_id = UNSET
elif isinstance(self.payer_id, UUID):
payer_id = str(self.payer_id)
else:
payer_id = self.payer_id
payer_type: Union[None, Unset, int]
if isinstance(self.payer_type, Unset):
payer_type = UNSET
elif isinstance(self.payer_type, PayerTypeEnum):
payer_type = self.payer_type.value
else:
payer_type = self.payer_type
delay = self.delay
field_dict: dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"uuid": uuid,
"payee_id": payee_id,
"payee_type": payee_type,
"created_at": created_at,
"state": state,
"amount": amount,
"updated_at": updated_at,
"details": details,
"role": role,
}
)
if payer_id is not UNSET:
field_dict["payer_id"] = payer_id
if payer_type is not UNSET:
field_dict["payer_type"] = payer_type
if delay is not UNSET:
field_dict["delay"] = delay
return field_dict
@classmethod
def from_dict(cls: type[T], src_dict: dict[str, Any]) -> T:
d = src_dict.copy()
uuid = UUID(d.pop("uuid"))
payee_id = UUID(d.pop("payee_id"))
payee_type = UUID(d.pop("payee_type"))
def _parse_created_at(data: object) -> Union[None, datetime.datetime]:
if data is None:
return data
try:
if not isinstance(data, str):
raise TypeError()
created_at_type_0 = isoparse(data)
return created_at_type_0
except: # noqa: E722
pass
return cast(Union[None, datetime.datetime], data)
created_at = _parse_created_at(d.pop("created_at"))
state = StateEnum(d.pop("state"))
amount = d.pop("amount")
def _parse_updated_at(data: object) -> Union[None, datetime.datetime]:
if data is None:
return data
try:
if not isinstance(data, str):
raise TypeError()
updated_at_type_0 = isoparse(data)
return updated_at_type_0
except: # noqa: E722
pass
return cast(Union[None, datetime.datetime], data)
updated_at = _parse_updated_at(d.pop("updated_at"))
details = d.pop("details")
role = d.pop("role")
def _parse_payer_id(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()
payer_id_type_0 = UUID(data)
return payer_id_type_0
except: # noqa: E722
pass
return cast(Union[None, UUID, Unset], data)
payer_id = _parse_payer_id(d.pop("payer_id", UNSET))
def _parse_payer_type(data: object) -> Union[None, PayerTypeEnum, Unset]:
if data is None:
return data
if isinstance(data, Unset):
return data
try:
if not isinstance(data, int):
raise TypeError()
payer_type_type_0 = PayerTypeEnum(data)
return payer_type_type_0
except: # noqa: E722
pass
return cast(Union[None, PayerTypeEnum, Unset], data)
payer_type = _parse_payer_type(d.pop("payer_type", UNSET))
delay = d.pop("delay", UNSET)
user_transaction = cls(
uuid=uuid,
payee_id=payee_id,
payee_type=payee_type,
created_at=created_at,
state=state,
amount=amount,
updated_at=updated_at,
details=details,
role=role,
payer_id=payer_id,
payer_type=payer_type,
delay=delay,
)
user_transaction.additional_properties = d
return user_transaction
@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
gooyal_wallet_client/py.typed Normal file
View file

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

46
gooyal_wallet_client/types.py Normal file
View file

@ -0,0 +1,46 @@
"""Contains some shared types for properties"""
from collections.abc import MutableMapping
from http import HTTPStatus
from typing import BinaryIO, Generic, Literal, Optional, TypeVar
from attrs import define
class Unset:
def __bool__(self) -> Literal[False]:
return False
UNSET: Unset = Unset()
FileJsonType = tuple[Optional[str], BinaryIO, Optional[str]]
@define
class File:
"""Contains information for file uploads"""
payload: BinaryIO
file_name: Optional[str] = None
mime_type: Optional[str] = None
def to_tuple(self) -> FileJsonType:
"""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", "FileJsonType", "Response", "Unset"]

3
requirements.in
View file

@ -16,4 +16,5 @@ django-oauth-toolkit
redis
celery
django-redis
django_minio_backend
django_minio_backend
httpx

18
requirements.txt
View file

@ -6,6 +6,8 @@
#
amqp==5.3.1
# via kombu
anyio==4.9.0
# via httpx
argon2-cffi==25.1.0
# via minio
argon2-cffi-bindings==21.2.0
@ -26,6 +28,8 @@ celery==5.5.3
# via -r requirements.in
certifi==2025.7.14
# via
# httpcore
# httpx
# minio
# requests
cffi==1.17.1
@ -91,8 +95,17 @@ greenlet==3.2.3
# via gevent
gunicorn==23.0.0
# via -r requirements.in
h11==0.16.0
# via httpcore
httpcore==1.0.9
# via httpx
httpx==0.28.1
# via -r requirements.in
idna==3.10
# via requests
# via
# anyio
# httpx
# requests
inflection==0.5.1
# via drf-spectacular
jalali-core==1.0.0
@ -153,10 +166,13 @@ rpds-py==0.26.0
# referencing
six==1.17.0
# via python-dateutil
sniffio==1.3.1
# via anyio
sqlparse==0.5.3
# via django
typing-extensions==4.14.1
# via
# anyio
# jwcrypto
# minio
# psycopg

0
utils/clients/__init__.py Normal file
View file

175
utils/clients/wallet_client.py Normal file
View file

@ -0,0 +1,175 @@
import time
import requests
from django.conf import settings
from django.core.cache import cache
from django.db import models
from httpx import Request
from gooyal_wallet_client import AuthenticatedClient
from gooyal_wallet_client.models import MyWallet, UserTransaction, ApplicationDeposit, ApplicationWithdraw
from gooyal_wallet_client.api.wallet import (wallet_api_application_deposit_create,
wallet_api_application_deposit_verify_retrieve,
wallet_api_application_withdraw_create,
wallet_api_application_withdraw_verify_retrieve)
import logging
# logging.basicConfig(
# format="%(levelname)s [%(asctime)s] %(name)s - %(message)s",
# datefmt="%Y-%m-%d %H:%M:%S",
# level=logging.DEBUG
# )
logger = logging.getLogger('ads.utils.wallet_client')
class ContentTypeChoices(models.IntegerChoices):
IMAGE = 1, 'image'
VOICE = 2, 'voice'
VIDEO = 3, 'video'
def login_as_client_credentials():
access_token = cache.get('ads_access_token')
# expires_in = self.auth_data.get('expires_in')
# created_at = self.auth_data.get('created_at')
#
# if access_token and expires_in > time.time() - created_at:
# return self.auth_data
if access_token:
logger.debug('token available in cache')
return access_token
logger.debug('no token available in cache')
data = {
"grant_type": 'client_credentials',
"scope": settings.SCOPES,
}
auth = (settings.CLIENT_ID, settings.CLIENT_SECRET)
response = requests.post(f'{settings.BASE_OAUTH2_PROVIDER_PRIVATE_URL}/token/',
data=data,
auth=auth)
logger.debug(response.content)
auth_data = response.json()
if 'access_token' in auth_data:
auth_data['created_at'] = time.time()
cache.set('ads_access_token', auth_data, timeout=auth_data.get('expires_in') - 5)
return auth_data
def log_request(request: Request):
logger.debug(f'authorization: {request.headers["authorization"]}')
logger.debug(
f"Request event hook: {request.method} {request.url} {request.method} {request.headers} - Waiting for response")
def log_response(response):
request = response.request
logger.debug(f"Response event hook: {request.method} {request.url} - Status {response.status_code}")
# logger.debug(f"Response event hook: {response.read()}")
def get_client():
access_token = login_as_client_credentials()['access_token']
client = AuthenticatedClient(base_url="https://wallet.gooyal.com", token=access_token,
httpx_args={"event_hooks": {"request": [log_request], "response": [log_response]}},
)
return client
def deposit_to_user_wallet_submit(data):
class Tmp():
def to_dict(self):
return data
deposit_data = Tmp()
client = get_client()
response: ApplicationDeposit = wallet_api_application_deposit_create.sync(client=client, body=deposit_data)
return response
def deposit_to_user_wallet_verify(uuid):
client = get_client()
response: ApplicationDeposit = wallet_api_application_deposit_verify_retrieve.sync(uuid=uuid, client=client)
return response
def withdraw_from_user_wallet_submit(data):
class Tmp():
def to_dict(self):
return data
withdraw_data = Tmp()
client = get_client()
response: ApplicationWithdraw = wallet_api_application_withdraw_create.sync(client=client, body=withdraw_data)
return response
def withdraw_from_user_wallet_verify(uuid):
client = get_client()
response: ApplicationWithdraw = wallet_api_application_withdraw_verify_retrieve.sync(uuid=uuid, client=client)
return response
# def deposit_to_user_wallet_submit(data):
# import uuid as sys_uuid
# class Tmp():
# uuid = sys_uuid.uuid4()
#
#
# data = Tmp()
#
# return data
#
#
# def deposit_to_user_wallet_verify(uuid):
# import uuid as sys_uuid
# class Tmp():
# uuid = sys_uuid.uuid4()
# state = 5
#
# data = Tmp()
#
# return data
#
#
#
# def withdraw_from_user_wallet_submit(data):
# import uuid as sys_uuid
# class Tmp():
# uuid = sys_uuid.uuid4()
#
# data = Tmp()
#
# return data
#
#
#
# def withdraw_from_user_wallet_verify(uuid):
# import uuid as sys_uuid
# class Tmp():
# uuid = sys_uuid.uuid4()
# state = 5
#
# data = Tmp()
#
# return data
#
sample_data = {
"uuid": "3fa85f64-5717-4562-b3fc-2c963f66afb1",
"payee_id": "5d66d02e-6087-4b0a-bb0c-291d5c11755a",
"payee_type": 1,
"amount": 2000,
"details": "string",
}