From c2f768769ad0d018383e5cdba0c1359b6f955c48 Mon Sep 17 00:00:00 2001 From: jab3z Date: Tue, 17 Jun 2025 23:11:06 +0300 Subject: [PATCH] Fix async request handler import --- src/ottu/__init__.py | 2 + src/ottu/cards_async.py | 97 +++++ src/ottu/decorators.py | 12 + src/ottu/ottu_async.py | 369 +++++++++++++++++++ src/ottu/request_async.py | 45 +++ src/ottu/session_async.py | 722 ++++++++++++++++++++++++++++++++++++++ 6 files changed, 1247 insertions(+) create mode 100644 src/ottu/cards_async.py create mode 100644 src/ottu/ottu_async.py create mode 100644 src/ottu/request_async.py create mode 100644 src/ottu/session_async.py diff --git a/src/ottu/__init__.py b/src/ottu/__init__.py index d9da86c..46e5af5 100644 --- a/src/ottu/__init__.py +++ b/src/ottu/__init__.py @@ -1,5 +1,7 @@ from .ottu import Ottu +from .ottu_async import OttuAsync __all__ = [ "Ottu", + "OttuAsync", ] diff --git a/src/ottu/cards_async.py b/src/ottu/cards_async.py new file mode 100644 index 0000000..1f89260 --- /dev/null +++ b/src/ottu/cards_async.py @@ -0,0 +1,97 @@ +import typing + +from . import urls +from .enums import HTTPMethod +from .request import OttuPYResponse +from .utils.helpers import remove_empty_values + +if typing.TYPE_CHECKING: + from .ottu_async import OttuAsync + + +class AsyncCard: + def __init__(self, ottu: "OttuAsync"): + self.ottu = ottu + + async def _get_cards( + self, + customer_id: typing.Optional[str] = None, + pg_codes: typing.Optional[typing.List] = None, + agreement_id: typing.Optional[str] = None, + ) -> OttuPYResponse: + if customer_id is None: + customer_id = self.ottu.customer_id + payload = { + "type": self.ottu.env_type, + "customer_id": customer_id, + "pg_codes": pg_codes, + "agreement_id": agreement_id, + } + payload = remove_empty_values(payload) + return await self.ottu.send_request( + path=urls.USER_CARDS, + method=HTTPMethod.POST, + json=payload, + ) + + async def get_cards( + self, + customer_id: typing.Optional[str] = None, + pg_codes: typing.Optional[typing.List] = None, + agreement_id: typing.Optional[str] = None, + ) -> dict: + ottu_py_response = await self._get_cards( + customer_id=customer_id, + pg_codes=pg_codes, + agreement_id=agreement_id, + ) + return ottu_py_response.as_dict() + + async def list( + self, + customer_id: typing.Optional[str] = None, + pg_codes: typing.Optional[typing.List] = None, + agreement_id: typing.Optional[str] = None, + ) -> dict: + return await self.get_cards( + customer_id=customer_id, + pg_codes=pg_codes, + agreement_id=agreement_id, + ) + + async def get( + self, + customer_id: typing.Optional[str] = None, + pg_codes: typing.Optional[typing.List] = None, + agreement_id: typing.Optional[str] = None, + ) -> typing.Optional[dict]: + ottu_py_response = await self._get_cards( + customer_id=customer_id, + pg_codes=pg_codes, + agreement_id=agreement_id, + ) + if ottu_py_response.success: + try: + return ottu_py_response.response[0] + except IndexError: + pass + return None + + async def delete( + self, + token: str, + customer_id: typing.Optional[str] = None, + ) -> dict: + customer_id = customer_id or self.ottu.customer_id + ottu_py_response = await self.ottu.send_request( + path=f"{urls.USER_CARDS}{token}/", + method=HTTPMethod.DELETE, + params={ + "customer_id": customer_id, + "type": self.ottu.env_type, + }, + ) + return ottu_py_response.as_dict() + + def __repr__(self): + return f"AsyncCard({self.ottu.session.customer_id})" diff --git a/src/ottu/decorators.py b/src/ottu/decorators.py index 3270c5a..54b5bbf 100644 --- a/src/ottu/decorators.py +++ b/src/ottu/decorators.py @@ -11,3 +11,15 @@ def wrapper(*args, **kwargs): return e.as_dict() return wrapper + + +def async_interruption_handler(func): + """Async decorator to handle keyboard interruption.""" + + async def wrapper(*args, **kwargs): + try: + return await func(*args, **kwargs) + except APIInterruptError as e: + return e.as_dict() + + return wrapper diff --git a/src/ottu/ottu_async.py b/src/ottu/ottu_async.py new file mode 100644 index 0000000..27f3e77 --- /dev/null +++ b/src/ottu/ottu_async.py @@ -0,0 +1,369 @@ +import typing + +import httpx +from httpx import Auth + +from . import urls +from .cards_async import AsyncCard +from .enums import HTTPMethod, TxnType +from .request import OttuPYResponse +from .request_async import AsyncRequestResponseHandler +from .session_async import AsyncSession +from .utils.helpers import remove_empty_values + + +class OttuAsync: + _session: typing.Optional[AsyncSession] = None + _card: typing.Optional[AsyncCard] = None + default_timeout: int = 30 + session_cls: typing.Type[AsyncSession] = AsyncSession + request_response_handler: typing.Type[AsyncRequestResponseHandler] = ( + AsyncRequestResponseHandler + ) + + def __init__( + self, + merchant_id: str, + auth: Auth, + customer_id: typing.Optional[str] = None, + is_sandbox: bool = True, + timeout: typing.Optional[int] = None, + ) -> None: + self.merchant_id = merchant_id + self.host_url = f"https://{merchant_id}" + self.auth = auth + self.customer_id = customer_id + self.is_sandbox = is_sandbox + self.env_type = "sandbox" if is_sandbox else "production" + self.timeout = timeout or self.default_timeout + + # Other initializations + self.request_session = self.__create_session() + + def __create_session(self) -> httpx.AsyncClient: + return httpx.AsyncClient(auth=self.auth) + + async def send_request( + self, + path: str, + method: str, + **request_params, + ) -> OttuPYResponse: + handler = self.request_response_handler( + session=self.request_session, + method=method, + url=f"{self.host_url}{path}", + timeout=self.timeout, + **request_params, + ) + return await handler.process() + + # Core Methods + + @property + def session(self): + if self._session is None: + self._session = self.session_cls(ottu=self) + return self._session + + def _update_session(self, session: AsyncSession) -> None: + self._session = session + + async def checkout( + self, + *, + txn_type: TxnType, + amount: str, + currency_code: str, + pg_codes: list[str], + payment_type: str = "one_off", + customer_id: typing.Optional[str] = None, + customer_email: typing.Optional[str] = None, + customer_phone: typing.Optional[str] = None, + customer_first_name: typing.Optional[str] = None, + customer_last_name: typing.Optional[str] = None, + agreement: typing.Optional[dict] = None, + card_acceptance_criteria: typing.Optional[dict] = None, + attachment: typing.Optional[str] = None, + billing_address: typing.Optional[dict] = None, + due_datetime: typing.Optional[str] = None, + email_recipients: typing.Optional[list[str]] = None, + expiration_time: typing.Optional[str] = None, + extra: typing.Optional[dict] = None, + generate_qr_code: typing.Optional[bool] = None, + language: typing.Optional[str] = None, + mode: typing.Optional[str] = None, + notifications: typing.Optional[dict] = None, + order_no: typing.Optional[str] = None, + product_type: typing.Optional[str] = None, + redirect_url: typing.Optional[str] = None, + shopping_address: typing.Optional[dict] = None, + shortify_attachment_url: typing.Optional[bool] = None, + shortify_checkout_url: typing.Optional[bool] = None, + vendor_name: typing.Optional[str] = None, + webhook_url: typing.Optional[str] = None, + include_sdk_setup_preload: typing.Optional[bool] = None, + **kwargs, + ) -> dict: + """ + a proxy method to `Session.create(...)` + """ + return await self.session.create( + txn_type=txn_type, + amount=amount, + currency_code=currency_code, + pg_codes=pg_codes, + payment_type=payment_type, + customer_id=customer_id, + customer_email=customer_email, + customer_phone=customer_phone, + customer_first_name=customer_first_name, + customer_last_name=customer_last_name, + agreement=agreement, + card_acceptance_criteria=card_acceptance_criteria, + attachment=attachment, + billing_address=billing_address, + due_datetime=due_datetime, + email_recipients=email_recipients, + expiration_time=expiration_time, + extra=extra, + generate_qr_code=generate_qr_code, + language=language, + mode=mode, + notifications=notifications, + order_no=order_no, + product_type=product_type, + redirect_url=redirect_url, + shopping_address=shopping_address, + shortify_attachment_url=shortify_attachment_url, + shortify_checkout_url=shortify_checkout_url, + vendor_name=vendor_name, + webhook_url=webhook_url, + include_sdk_setup_preload=include_sdk_setup_preload, + **kwargs, + ) + + def auto_debit( + self, + token: str, + session_id: str, + ): + return await self.session.auto_debit(token=token, session_id=session_id) + + @property + def cards(self) -> Card: + if self._card is None: + self._card = Card(ottu=self) + return self._card + + async def checkout_autoflow( + self, + *, + txn_type: TxnType, + amount: str, + currency_code: str, + payment_type: str = "one_off", + customer_id: typing.Optional[str] = None, + customer_email: typing.Optional[str] = None, + customer_phone: typing.Optional[str] = None, + customer_first_name: typing.Optional[str] = None, + customer_last_name: typing.Optional[str] = None, + agreement: typing.Optional[dict] = None, + card_acceptance_criteria: typing.Optional[dict] = None, + attachment: typing.Optional[str] = None, + billing_address: typing.Optional[dict] = None, + due_datetime: typing.Optional[str] = None, + email_recipients: typing.Optional[list[str]] = None, + expiration_time: typing.Optional[str] = None, + extra: typing.Optional[dict] = None, + generate_qr_code: typing.Optional[bool] = None, + language: typing.Optional[str] = None, + mode: typing.Optional[str] = None, + notifications: typing.Optional[dict] = None, + order_no: typing.Optional[str] = None, + product_type: typing.Optional[str] = None, + redirect_url: typing.Optional[str] = None, + shopping_address: typing.Optional[dict] = None, + shortify_attachment_url: typing.Optional[bool] = None, + shortify_checkout_url: typing.Optional[bool] = None, + vendor_name: typing.Optional[str] = None, + webhook_url: typing.Optional[str] = None, + include_sdk_setup_preload: typing.Optional[bool] = None, + checkout_extra_args: typing.Optional[dict] = None, + ): + return await self.session.checkout_autoflow( + txn_type=txn_type, + amount=amount, + currency_code=currency_code, + payment_type=payment_type, + customer_id=customer_id, + customer_email=customer_email, + customer_phone=customer_phone, + customer_first_name=customer_first_name, + customer_last_name=customer_last_name, + agreement=agreement, + card_acceptance_criteria=card_acceptance_criteria, + attachment=attachment, + billing_address=billing_address, + due_datetime=due_datetime, + email_recipients=email_recipients, + expiration_time=expiration_time, + extra=extra, + generate_qr_code=generate_qr_code, + language=language, + mode=mode, + notifications=notifications, + order_no=order_no, + product_type=product_type, + redirect_url=redirect_url, + shopping_address=shopping_address, + shortify_attachment_url=shortify_attachment_url, + shortify_checkout_url=shortify_checkout_url, + vendor_name=vendor_name, + webhook_url=webhook_url, + include_sdk_setup_preload=include_sdk_setup_preload, + checkout_extra_args=checkout_extra_args, + ) + + async def auto_debit_autoflow( + self, + *, + txn_type: TxnType, + amount: str, + currency_code: str, + customer_id: str, + agreement: dict, + pg_codes: typing.Optional[list[str]] = None, + customer_email: typing.Optional[str] = None, + customer_phone: typing.Optional[str] = None, + customer_first_name: typing.Optional[str] = None, + customer_last_name: typing.Optional[str] = None, + card_acceptance_criteria: typing.Optional[dict] = None, + attachment: typing.Optional[str] = None, + billing_address: typing.Optional[dict] = None, + due_datetime: typing.Optional[str] = None, + email_recipients: typing.Optional[list[str]] = None, + expiration_time: typing.Optional[str] = None, + extra: typing.Optional[dict] = None, + generate_qr_code: typing.Optional[bool] = None, + language: typing.Optional[str] = None, + mode: typing.Optional[str] = None, + notifications: typing.Optional[dict] = None, + order_no: typing.Optional[str] = None, + product_type: typing.Optional[str] = None, + redirect_url: typing.Optional[str] = None, + shopping_address: typing.Optional[dict] = None, + shortify_attachment_url: typing.Optional[bool] = None, + shortify_checkout_url: typing.Optional[bool] = None, + vendor_name: typing.Optional[str] = None, + webhook_url: typing.Optional[str] = None, + include_sdk_setup_preload: typing.Optional[bool] = None, + checkout_extra_args: typing.Optional[dict] = None, + token: typing.Optional[str] = None, + ): + return await self.session.auto_debit_autoflow( + txn_type=txn_type, + amount=amount, + currency_code=currency_code, + customer_id=customer_id, + pg_codes=pg_codes, + customer_email=customer_email, + customer_phone=customer_phone, + customer_first_name=customer_first_name, + customer_last_name=customer_last_name, + agreement=agreement, + card_acceptance_criteria=card_acceptance_criteria, + attachment=attachment, + billing_address=billing_address, + due_datetime=due_datetime, + email_recipients=email_recipients, + expiration_time=expiration_time, + extra=extra, + generate_qr_code=generate_qr_code, + language=language, + mode=mode, + notifications=notifications, + order_no=order_no, + product_type=product_type, + redirect_url=redirect_url, + shopping_address=shopping_address, + shortify_attachment_url=shortify_attachment_url, + shortify_checkout_url=shortify_checkout_url, + vendor_name=vendor_name, + webhook_url=webhook_url, + include_sdk_setup_preload=include_sdk_setup_preload, + checkout_extra_args=checkout_extra_args, + token=token, + ) + + async def raw( + self, + method: str, + path: str, + headers: typing.Optional[dict] = None, + **kwargs, + ): + """ + To send any sort of http requests to the server. + method: str - HTTP method name. + Eg: GET, POST, PUT, DELETE, etc. + path: str - The path of the request. + Eg: /b/api/v1/dashboard/statistics + headers: dict - Optional headers to be sent with the request. + kwargs: dict - Optional parameters to be sent with the request. + Supports all the parameters that `httpx.Client.send` supports. + """ + return await self.send_request( + path=path, + method=method, + headers=headers, + **kwargs, + ) + + def get_payment_methods( + self, + plugin, + currencies: typing.Optional[list[str]] = None, + customer_id: typing.Optional[str] = None, + operation: typing.Optional[str] = None, + tokenizable: bool = False, + pg_names: typing.Optional[list[str]] = None, + ) -> dict: + return ( + await self._get_payment_methods( + plugin=plugin, + currencies=currencies, + customer_id=customer_id, + operation=operation, + tokenizable=tokenizable, + pg_names=pg_names, + ) + ).as_dict() + + async def _get_payment_methods( + self, + plugin, + currencies: typing.Optional[list[str]] = None, + customer_id: typing.Optional[str] = None, + operation: typing.Optional[str] = None, + tokenizable: bool = False, + pg_names: typing.Optional[list[str]] = None, + ) -> OttuPYResponse: + payload = { + "plugin": plugin, + "currencies": currencies, + "customer_id": customer_id, + "operation": operation, + "tokenizable": tokenizable, + "pg_names": pg_names, + "type": "sandbox" if self.is_sandbox else "production", + } + payload = remove_empty_values(payload) + return await self.send_request( + path=urls.PAYMENT_METHODS, + method=HTTPMethod.POST, + json=payload, + ) + + def __repr__(self): + return f"Ottu({self.merchant_id})" diff --git a/src/ottu/request_async.py b/src/ottu/request_async.py new file mode 100644 index 0000000..fdc19d7 --- /dev/null +++ b/src/ottu/request_async.py @@ -0,0 +1,45 @@ +import logging +from json import JSONDecodeError +from urllib.parse import urlparse + +import httpx + +from .mixins import ResponseMixin +from .request import OttuPYResponse, RequestResponseHandler + +logger = logging.getLogger("ottu-py") + + +class AsyncRequestResponseHandler(RequestResponseHandler): + def __init__(self, session: httpx.AsyncClient, method: str, url: str, **kwargs): + super().__init__(session=session, method=method, url=url, **kwargs) + + async def _process(self) -> OttuPYResponse: + try: + logger.info( + f"Sending {self.method} request to {self.url} with args {self.kwargs}" + ) + response = await self.session.request( + method=self.method, + url=self.url, + **self.kwargs, + ) + return self.process_response(response) + except httpx.HTTPError as exc: + return self.process_httpx_error(exc) + except Exception as exc: + return self.process_unknown_error(exc) + + async def process(self) -> OttuPYResponse: + response = await self._process() + if response.success: + logger.info( + f"Received {response.status_code} response " + f"from {self.url} with args {self.kwargs}" + ) + else: + logger.error( + f"Received {response.status_code} response " + f"from {self.url} with args {self.kwargs}. Error: {response.error}" + ) + return response diff --git a/src/ottu/session_async.py b/src/ottu/session_async.py new file mode 100644 index 0000000..41f0c2f --- /dev/null +++ b/src/ottu/session_async.py @@ -0,0 +1,722 @@ +import logging +import typing + +from .decorators import async_interruption_handler +from .enums import HTTPMethod, TxnType +from .errors import APIInterruptError, ValidationError +from .mixins import AsDictMixin +from .request import OttuPYResponse +from .utils.dataclasses import dynamic_dataclass +from .utils.helpers import remove_empty_values + +if typing.TYPE_CHECKING: + from .ottu import Ottu + +logger = logging.getLogger("ottu-py") + + +@dynamic_dataclass +class PaymentMethod(AsDictMixin): + code: typing.Optional[str] = None + name: typing.Optional[str] = None + pg: typing.Optional[str] = None + type: typing.Optional[str] = None + amount: typing.Optional[str] = None + currency_code: typing.Optional[str] = None + fee: typing.Optional[str] = None + fee_description: typing.Optional[str] = None + icon: typing.Optional[str] = None + flow: typing.Optional[str] = None + redirect_url: typing.Optional[str] = None + + def __repr__(self): + return f"PaymentMethod({self.code or '######'})" + + +class AsyncSession: + url_session_create = "/b/checkout/v1/pymt-txn/" + url_ops = "/b/pbl/v2/operation/" + url_auto_debit = "/b/pbl/v2/auto-debit/" + + amount: typing.Optional[str] = None + attachment: typing.Optional[str] = None + attachment_short_url: typing.Optional[str] = None + billing_address: typing.Optional[dict] = None + checkout_short_url: typing.Optional[str] = None + checkout_url: typing.Optional[str] = None + currency_code: typing.Optional[str] = None + customer_email: typing.Optional[str] = None + customer_first_name: typing.Optional[str] = None + customer_last_name: typing.Optional[str] = None + customer_id: typing.Optional[str] = None + customer_phone: typing.Optional[str] = None + due_datetime: typing.Optional[str] = None + email_recipients: typing.Optional[list[str]] = None + expiration_time: typing.Optional[str] = None + extra: typing.Optional[dict] = None + initiator_id: typing.Optional[int] = None + language: typing.Optional[str] = None + mode: typing.Optional[str] = None + notifications: typing.Optional[dict] = None + operation: typing.Optional[str] = None + order_no: typing.Optional[str] = None + payment_methods: typing.Optional[list[PaymentMethod]] = None + pg_codes: typing.Optional[list[str]] = None + qr_code_url: typing.Optional[str] = None + redirect_url: typing.Optional[str] = None + session_id: typing.Optional[str] = None + shipping_address: typing.Optional[dict] = None + state: typing.Optional[str] = None + type: typing.Optional[str] = None + vendor_name: typing.Optional[str] = None + webhook_url: typing.Optional[str] = None + + def __init__(self, ottu: "Ottu", **data): + self.ottu = ottu + for field, value in data.items(): + setattr(self, field, value) + + payment_methods = getattr(self, "payment_methods", []) + if payment_methods: + self.payment_methods = [ + PaymentMethod( + **payment_method, + ) + for payment_method in payment_methods + ] + + def __repr__(self): + return f"Session({self.session_id or '######'})" + + def __bool__(self): + return bool(self.session_id) + + def as_dict(self): + fields = [ + "amount", + "attachment", + "attachment_short_url", + "billing_address", + "checkout_short_url", + "checkout_url", + "currency_code", + "customer_email", + "customer_first_name", + "customer_last_name", + "customer_id", + "customer_phone", + "due_datetime", + "email_recipients", + "expiration_time", + "extra", + "initiator_id", + "language", + "mode", + "notifications", + "operation", + "order_no", + "payment_methods", + "pg_codes", + "qr_code_url", + "redirect_url", + "session_id", + "shipping_address", + "state", + "type", + "vendor_name", + "webhook_url", + ] + return remove_empty_values( + {field: getattr(self, field, "") for field in fields}, + ) + + def __path_to_file(self, path: str) -> typing.Tuple[str, bytes]: + with open(path, "rb") as f: + content = f.read() + name = f.name or "attachment.pdf" + return name, content + + async def create( + self, + *, + txn_type: TxnType, + amount: str, + currency_code: str, + pg_codes: list[str], + payment_type: str = "one_off", + customer_id: typing.Optional[str] = None, + customer_email: typing.Optional[str] = None, + customer_phone: typing.Optional[str] = None, + customer_first_name: typing.Optional[str] = None, + customer_last_name: typing.Optional[str] = None, + agreement: typing.Optional[dict] = None, + card_acceptance_criteria: typing.Optional[dict] = None, + attachment: typing.Optional[str] = None, + billing_address: typing.Optional[dict] = None, + due_datetime: typing.Optional[str] = None, + email_recipients: typing.Optional[list[str]] = None, + expiration_time: typing.Optional[str] = None, + extra: typing.Optional[dict] = None, + generate_qr_code: typing.Optional[bool] = None, + language: typing.Optional[str] = None, + mode: typing.Optional[str] = None, + notifications: typing.Optional[dict] = None, + order_no: typing.Optional[str] = None, + product_type: typing.Optional[str] = None, + redirect_url: typing.Optional[str] = None, + shopping_address: typing.Optional[dict] = None, + shortify_attachment_url: typing.Optional[bool] = None, + shortify_checkout_url: typing.Optional[bool] = None, + vendor_name: typing.Optional[str] = None, + webhook_url: typing.Optional[str] = None, + include_sdk_setup_preload: typing.Optional[bool] = None, + **kwargs, + ) -> dict: + """ + Creates a new checkout session. + :param txn_type: Transaction type + :param amount: Amount + :param currency_code: Currency code + :param pg_codes: Payment gateway codes + :param customer_id: Customer ID + :param customer_email: Customer email + :param customer_phone: Customer phone + :param customer_first_name: Customer first name + :param customer_last_name: Customer last name + :param agreement: Agreement + :param card_acceptance_criteria: Card acceptance criteria + :param attachment: Path to attachment + :param billing_address: Billing address + :param due_datetime: Due datetime + :param email_recipients: Email recipients + :param expiration_time: Expiration time + :param extra: Extra + :param generate_qr_code: Generate QR code + :param language: Language + :param mode: Mode + :param notifications: Notifications + :param order_no: Order number + :param product_type: Product type + :param redirect_url: Redirect URL + :param shopping_address: Shopping address + :param shortify_attachment_url: Shortify attachment URL + :param shortify_checkout_url: Shortify checkout URL + :param vendor_name: Vendor name + :param webhook_url: Webhook URL + :param include_sdk_setup_preload: Include SDK setup preload + :param kwargs: Additional arguments supported by the API + :return: Session + """ + if kwargs: + msg = ( + f"The following arguments are not " + f"supported by the SDK: {', '.join(kwargs.keys())}" + ) + logger.warning(msg) + + customer_id = customer_id or self.ottu.customer_id + payload = { + "type": txn_type.value, + "amount": amount, + "currency_code": currency_code, + "pg_codes": pg_codes, + "payment_type": payment_type, + "customer_id": customer_id, + "customer_email": customer_email, + "customer_phone": customer_phone, + "customer_first_name": customer_first_name, + "customer_last_name": customer_last_name, + "agreement": agreement, + "card_acceptance_criteria": card_acceptance_criteria, + "billing_address": billing_address, + "due_datetime": due_datetime, + "email_recipients": email_recipients, + "expiration_time": expiration_time, + "extra": extra, + "generate_qr_code": generate_qr_code, + "language": language, + "mode": mode, + "notifications": notifications, + "order_no": order_no, + "product_type": product_type, + "redirect_url": redirect_url, + "shopping_address": shopping_address, + "shortify_attachment_url": shortify_attachment_url, + "shortify_checkout_url": shortify_checkout_url, + "vendor_name": vendor_name, + "webhook_url": webhook_url, + "include_sdk_setup_preload": include_sdk_setup_preload, + } + payload = remove_empty_values(payload) + payload.update(kwargs) # `kwargs` may contain `None` values + if attachment: + json_or_form = { + "data": payload, + "files": {"attachment": self.__path_to_file(path=attachment)}, + } + else: + json_or_form = { + "json": payload, + } + ottu_py_response = await self.ottu.send_request( + path=self.url_session_create, + method=HTTPMethod.POST, + **json_or_form, + ) + session = AsyncSession( + ottu=self.ottu, + **ottu_py_response.response, + ) + if ottu_py_response.success: + self.ottu._update_session(session) + return ottu_py_response.as_dict() + + async def retrieve(self, session_id: str) -> dict: + """ + Retrieves a checkout session. + :param session_id: Session ID + """ + ottu_py_response = await self.ottu.send_request( + path=f"{self.url_session_create}{session_id}", + method=HTTPMethod.GET, + ) + session = AsyncSession( + ottu=self.ottu, + **ottu_py_response.response, + ) + if ottu_py_response.success: + self.ottu._update_session(session) + return ottu_py_response.as_dict() + + async def refresh( + self, session_id: typing.Optional[str] = None + ) -> typing.Optional[dict]: + """ + Reloads the payment attributes from upstream by calling the `retrieve` method. + """ + session_id = session_id or self.session_id + if session_id: + response = await self.retrieve(session_id=session_id) + if response["success"]: + return response + return None + + async def update( + self, + *, + amount: typing.Optional[str] = None, + currency_code: typing.Optional[str] = None, + pg_codes: typing.Optional[list[str]] = None, + customer_id: typing.Optional[str] = None, + customer_email: typing.Optional[str] = None, + customer_phone: typing.Optional[str] = None, + customer_first_name: typing.Optional[str] = None, + customer_last_name: typing.Optional[str] = None, + attachment: typing.Optional[str] = None, + billing_address: typing.Optional[dict] = None, + due_datetime: typing.Optional[str] = None, + email_recipients: typing.Optional[list[str]] = None, + expiration_time: typing.Optional[str] = None, + extra: typing.Optional[dict] = None, + generate_qr_code: typing.Optional[bool] = None, + language: typing.Optional[str] = None, + mode: typing.Optional[str] = None, + notifications: typing.Optional[dict] = None, + order_no: typing.Optional[str] = None, + product_type: typing.Optional[str] = None, + redirect_url: typing.Optional[str] = None, + shopping_address: typing.Optional[dict] = None, + shortify_attachment_url: typing.Optional[bool] = None, + shortify_checkout_url: typing.Optional[bool] = None, + vendor_name: typing.Optional[str] = None, + webhook_url: typing.Optional[str] = None, + **kwargs, + ) -> dict: + payload = { + "amount": amount, + "currency_code": currency_code, + "pg_codes": pg_codes, + "customer_id": customer_id, + "customer_email": customer_email, + "customer_phone": customer_phone, + "customer_first_name": customer_first_name, + "customer_last_name": customer_last_name, + "billing_address": billing_address, + "due_datetime": due_datetime, + "email_recipients": email_recipients, + "expiration_time": expiration_time, + "extra": extra, + "generate_qr_code": generate_qr_code, + "language": language, + "mode": mode, + "notifications": notifications, + "order_no": order_no, + "product_type": product_type, + "redirect_url": redirect_url, + "shopping_address": shopping_address, + "shortify_attachment_url": shortify_attachment_url, + "shortify_checkout_url": shortify_checkout_url, + "vendor_name": vendor_name, + "webhook_url": webhook_url, + } + payload = remove_empty_values(payload) + payload.update(kwargs) # `kwargs` may contain `None` values + if attachment: + json_or_form = { + "data": payload, + "files": {"attachment": self.__path_to_file(path=attachment)}, + } + else: + json_or_form = { + "json": payload, + } + ottu_py_response = await self.ottu.send_request( + path=f"{self.url_session_create}{self.session_id}", + method=HTTPMethod.PATCH, + **json_or_form, + ) + session = AsyncSession(ottu=self.ottu, **ottu_py_response.response) + if ottu_py_response.success: + self.ottu._update_session(session) + return ottu_py_response.as_dict() + + async def auto_debit(self, token: str, session_id: str) -> dict: + payload = { + "session_id": session_id, + "token": token, + } + ottu_py_response = await self.ottu.send_request( + path=self.url_auto_debit, + method=HTTPMethod.POST, + json=payload, + ) + return ottu_py_response.as_dict() + + async def ops( + self, + operation: str, + order_id: typing.Optional[str] = None, + session_id: typing.Optional[str] = None, + amount: typing.Optional[str] = None, + headers: typing.Optional[dict] = None, + ) -> OttuPYResponse: + if session_id is None: + session_id = self.session_id + + if not session_id and not order_id: + raise ValidationError("session_id or order_id is required") + + payload = { + "session_id": session_id, + "order_no": order_id, + "operation": operation, + "amount": amount, + } + payload = remove_empty_values(payload) + return await self.ottu.send_request( + path=self.url_ops, + method=HTTPMethod.POST, + json=payload, + headers=headers, + ) + + async def cancel( + self, + order_id: typing.Optional[str] = None, + session_id: typing.Optional[str] = None, + ) -> dict: + ottu_py_response = self.ops( + operation="cancel", + order_id=order_id, + session_id=session_id, + ) + if ottu_py_response.success: + await self.refresh() + return ottu_py_response.as_dict() + + async def expire( + self, + order_id: typing.Optional[str] = None, + session_id: typing.Optional[str] = None, + ) -> dict: + ottu_py_response = self.ops( + operation="expire", + order_id=order_id, + session_id=session_id, + ) + if ottu_py_response.success: + await self.refresh() + return ottu_py_response.as_dict() + + async def delete( + self, + order_id: typing.Optional[str] = None, + session_id: typing.Optional[str] = None, + ) -> dict: + ottu_py_response = self.ops( + operation="delete", + order_id=order_id, + session_id=session_id, + ) + return ottu_py_response.as_dict() + + async def capture( + self, + order_id: typing.Optional[str] = None, + session_id: typing.Optional[str] = None, + amount: typing.Optional[str] = None, + tracking_key: typing.Optional[str] = None, + ) -> dict: + headers = None + if tracking_key: + headers = { + "Tracking-Key": tracking_key, + } + ottu_py_response = self.ops( + operation="capture", + order_id=order_id, + session_id=session_id, + amount=amount, + headers=headers, + ) + if ottu_py_response.success: + await self.refresh() + return ottu_py_response.as_dict() + + async def refund( + self, + order_id: typing.Optional[str] = None, + session_id: typing.Optional[str] = None, + amount: typing.Optional[str] = None, + tracking_key: typing.Optional[str] = None, + ) -> dict: + headers = None + if tracking_key: + headers = { + "Tracking-Key": tracking_key, + } + ottu_py_response = self.ops( + operation="refund", + order_id=order_id, + session_id=session_id, + amount=amount, + headers=headers, + ) + if ottu_py_response.success: + await self.refresh() + return ottu_py_response.as_dict() + + async def void( + self, + order_id: typing.Optional[str] = None, + session_id: typing.Optional[str] = None, + tracking_key: typing.Optional[str] = None, + ) -> dict: + headers = None + if tracking_key: + headers = { + "Tracking-Key": tracking_key, + } + ottu_py_response = self.ops( + operation="void", + order_id=order_id, + session_id=session_id, + headers=headers, + ) + if ottu_py_response.success: + await self.refresh() + return ottu_py_response.as_dict() + + async def get_pg_codes(self, plugin, currency, tokenizable=False) -> list: + if self.payment_methods: + return [pm.code for pm in self.payment_methods] + + response = self.ottu.get_payment_methods( + plugin=plugin, + currencies=[ + currency, + ], + tokenizable=tokenizable, + ) + if not response["success"]: + raise APIInterruptError(**response) + return [pm["code"] for pm in response["response"]["payment_methods"]] + + async def get_auto_debit_pg_codes(self, plugin, currency) -> list: + # There is no way to identify the + # cached payment method supports auto debit or not. + # So, we are calling the API again. + response = self.ottu.get_payment_methods( + plugin=plugin, + currencies=[ + currency, + ], + tokenizable=True, + ) + if not response["success"]: + raise APIInterruptError(**response) + return [pm["code"] for pm in response["response"]["payment_methods"]] + + def get_token_from_db(self, agreement, customer_id) -> str: + raise NotImplementedError("Please implement this method in your subclass") + + @async_interruption_handler + async def checkout_autoflow( + self, + *, + txn_type: TxnType, + amount: str, + currency_code: str, + payment_type: str = "one_off", + customer_id: typing.Optional[str] = None, + customer_email: typing.Optional[str] = None, + customer_phone: typing.Optional[str] = None, + customer_first_name: typing.Optional[str] = None, + customer_last_name: typing.Optional[str] = None, + agreement: typing.Optional[dict] = None, + card_acceptance_criteria: typing.Optional[dict] = None, + attachment: typing.Optional[str] = None, + billing_address: typing.Optional[dict] = None, + due_datetime: typing.Optional[str] = None, + email_recipients: typing.Optional[list[str]] = None, + expiration_time: typing.Optional[str] = None, + extra: typing.Optional[dict] = None, + generate_qr_code: typing.Optional[bool] = None, + language: typing.Optional[str] = None, + mode: typing.Optional[str] = None, + notifications: typing.Optional[dict] = None, + order_no: typing.Optional[str] = None, + product_type: typing.Optional[str] = None, + redirect_url: typing.Optional[str] = None, + shopping_address: typing.Optional[dict] = None, + shortify_attachment_url: typing.Optional[bool] = None, + shortify_checkout_url: typing.Optional[bool] = None, + vendor_name: typing.Optional[str] = None, + webhook_url: typing.Optional[str] = None, + include_sdk_setup_preload: typing.Optional[bool] = None, + checkout_extra_args: typing.Optional[dict] = None, + ): + pg_codes = await self.get_pg_codes(plugin=txn_type, currency=currency_code) + checkout_extra_args = checkout_extra_args or {} + return await self.create( + txn_type=txn_type, + amount=amount, + currency_code=currency_code, + pg_codes=pg_codes, + payment_type=payment_type, + customer_id=customer_id, + customer_email=customer_email, + customer_phone=customer_phone, + customer_first_name=customer_first_name, + customer_last_name=customer_last_name, + agreement=agreement, + card_acceptance_criteria=card_acceptance_criteria, + attachment=attachment, + billing_address=billing_address, + due_datetime=due_datetime, + email_recipients=email_recipients, + expiration_time=expiration_time, + extra=extra, + generate_qr_code=generate_qr_code, + language=language, + mode=mode, + notifications=notifications, + order_no=order_no, + product_type=product_type, + redirect_url=redirect_url, + shopping_address=shopping_address, + shortify_attachment_url=shortify_attachment_url, + shortify_checkout_url=shortify_checkout_url, + vendor_name=vendor_name, + webhook_url=webhook_url, + include_sdk_setup_preload=include_sdk_setup_preload, + **checkout_extra_args, + ) + + @async_interruption_handler + async def auto_debit_autoflow( + self, + *, + txn_type: TxnType, + amount: str, + currency_code: str, + customer_id: str, + agreement: dict, + pg_codes: typing.Optional[list[str]] = None, + customer_email: typing.Optional[str] = None, + customer_phone: typing.Optional[str] = None, + customer_first_name: typing.Optional[str] = None, + customer_last_name: typing.Optional[str] = None, + card_acceptance_criteria: typing.Optional[dict] = None, + attachment: typing.Optional[str] = None, + billing_address: typing.Optional[dict] = None, + due_datetime: typing.Optional[str] = None, + email_recipients: typing.Optional[list[str]] = None, + expiration_time: typing.Optional[str] = None, + extra: typing.Optional[dict] = None, + generate_qr_code: typing.Optional[bool] = None, + language: typing.Optional[str] = None, + mode: typing.Optional[str] = None, + notifications: typing.Optional[dict] = None, + order_no: typing.Optional[str] = None, + product_type: typing.Optional[str] = None, + redirect_url: typing.Optional[str] = None, + shopping_address: typing.Optional[dict] = None, + shortify_attachment_url: typing.Optional[bool] = None, + shortify_checkout_url: typing.Optional[bool] = None, + vendor_name: typing.Optional[str] = None, + webhook_url: typing.Optional[str] = None, + include_sdk_setup_preload: typing.Optional[bool] = None, + checkout_extra_args: typing.Optional[dict] = None, + token: typing.Optional[str] = None, + ): + """ + Completes the auto debit flow by automatically + identifying the "latest" payment method and the token. + """ + checkout_extra_args = checkout_extra_args or {} + if not token: + token = self.get_token_from_db(agreement=agreement, customer_id=customer_id) + if not pg_codes: + pg_codes = await self.get_auto_debit_pg_codes( + plugin=txn_type, + currency=currency_code, + ) + checkout_response = await self.create( + txn_type=txn_type, + amount=amount, + currency_code=currency_code, + pg_codes=pg_codes, + payment_type="auto_debit", + customer_id=customer_id, + customer_email=customer_email, + customer_phone=customer_phone, + customer_first_name=customer_first_name, + customer_last_name=customer_last_name, + agreement=agreement, + card_acceptance_criteria=card_acceptance_criteria, + attachment=attachment, + billing_address=billing_address, + due_datetime=due_datetime, + email_recipients=email_recipients, + expiration_time=expiration_time, + extra=extra, + generate_qr_code=generate_qr_code, + language=language, + mode=mode, + notifications=notifications, + order_no=order_no, + product_type=product_type, + redirect_url=redirect_url, + shopping_address=shopping_address, + shortify_attachment_url=shortify_attachment_url, + shortify_checkout_url=shortify_checkout_url, + vendor_name=vendor_name, + webhook_url=webhook_url, + include_sdk_setup_preload=include_sdk_setup_preload, + **checkout_extra_args, + ) + if not checkout_response["success"]: + raise APIInterruptError(**checkout_response) + session_id = checkout_response["response"]["session_id"] + return await self.auto_debit(token=token, session_id=session_id)