From e797765980e7f08f2b8a377e5ab738fbc742dbea Mon Sep 17 00:00:00 2001 From: Rostyslav Bohomaz Date: Thu, 9 Nov 2023 16:28:48 +0200 Subject: [PATCH] Refactor (#1) * add new types * move version from about to init * redesign api functions * add json encoder tests * add pytest ini file * update notebook * move test server to source * Update Liqpay client with new subscribe method * Update Liqpay types, exceptions, server, validation, and convert modules --- .vscode/tasks.json | 2 +- liqpy/__about__.py | 1 - liqpy/__init__.py | 1 + liqpy/api.py | 271 ++++++++++++++++++ liqpy/client.py | 545 ++++++++++++++++--------------------- liqpy/constants.py | 5 - liqpy/convert.py | 98 +++++++ liqpy/data.py | 150 ++++++++++ liqpy/exceptions.py | 198 ++++---------- liqpy/preprocess.py | 79 ++++++ {tests => liqpy}/server.py | 20 +- liqpy/testing.py | 8 +- liqpy/types.py | 160 ----------- liqpy/types/__init__.py | 6 + liqpy/types/action.py | 45 +++ liqpy/types/callback.py | 73 +++++ liqpy/types/common.py | 17 ++ liqpy/types/error.py | 143 ++++++++++ liqpy/types/post.py | 20 ++ liqpy/types/request.py | 122 +++++++++ liqpy/types/status.py | 40 +++ liqpy/util.py | 95 ------- liqpy/validation.py | 311 +++++++++++++++++++++ pyproject.toml | 2 +- pytest.ini | 0 readme.ipynb | 188 +++++++++---- tests/test_json_encoder.py | 67 +++++ 27 files changed, 1892 insertions(+), 775 deletions(-) delete mode 100644 liqpy/__about__.py create mode 100644 liqpy/api.py delete mode 100644 liqpy/constants.py create mode 100644 liqpy/convert.py create mode 100644 liqpy/data.py create mode 100644 liqpy/preprocess.py rename {tests => liqpy}/server.py (84%) delete mode 100644 liqpy/types.py create mode 100644 liqpy/types/__init__.py create mode 100644 liqpy/types/action.py create mode 100644 liqpy/types/callback.py create mode 100644 liqpy/types/common.py create mode 100644 liqpy/types/error.py create mode 100644 liqpy/types/post.py create mode 100644 liqpy/types/request.py create mode 100644 liqpy/types/status.py delete mode 100644 liqpy/util.py create mode 100644 liqpy/validation.py create mode 100644 pytest.ini create mode 100644 tests/test_json_encoder.py diff --git a/.vscode/tasks.json b/.vscode/tasks.json index e9a7162..5ec28c3 100644 --- a/.vscode/tasks.json +++ b/.vscode/tasks.json @@ -7,7 +7,7 @@ "command": "${command:python.interpreterPath}", "args": [ "-m", - "tests.server" + "liqpy.server" ], }, { diff --git a/liqpy/__about__.py b/liqpy/__about__.py deleted file mode 100644 index 493f741..0000000 --- a/liqpy/__about__.py +++ /dev/null @@ -1 +0,0 @@ -__version__ = "0.3.0" diff --git a/liqpy/__init__.py b/liqpy/__init__.py index e69de29..6a9beea 100644 --- a/liqpy/__init__.py +++ b/liqpy/__init__.py @@ -0,0 +1 @@ +__version__ = "0.4.0" diff --git a/liqpy/api.py b/liqpy/api.py new file mode 100644 index 0000000..c2cac80 --- /dev/null +++ b/liqpy/api.py @@ -0,0 +1,271 @@ +from typing import TYPE_CHECKING, Any, AnyStr, Optional, Unpack + +from functools import singledispatchmethod +from enum import Enum +from dataclasses import asdict + +from urllib.parse import urljoin +from base64 import b64encode, b64decode +from hashlib import sha1 +from json import loads, JSONEncoder + +from uuid import UUID +from decimal import Decimal +from datetime import date, datetime, UTC + +from .data import FiscalItem, DetailAddenda, SplitRule +from .preprocess import Preprocessor, BasePreprocessor +from .validation import Validator, BaseValidator + +if TYPE_CHECKING: + from requests import Session, Response + + from .types import LiqpayRequestDict + from .types.action import Action + from .types.post import Hooks, Proxies, Timeout, Verify, Cert + + +__all__ = ("Endpoint", "post", "sign", "encode", "decode", "request") + +URL = "https://www.liqpay.ua" +VERSION = 3 + +SENDER_KEYS = { + "sender_first_name", + "sender_last_name", + "sender_email", + "sender_address", + "sender_city", + "sender_country_code", + "sender_postal_code", + "sender_shipping_state", +} + +PRODUCT_KEYS = { + "product_category", + "product_description", + "product_name", + "product_url", +} + + +class Endpoint(Enum): + REQUEST: str = "/api/request" + CHECKOUT: str = f"/api/{VERSION}/checkout" + + def url(self) -> str: + return urljoin(URL, self.value) + + +class LiqPayJSONEncoder(JSONEncoder): + date_fmt = r"%Y-%m-%d %H:%M:%S" + + def __init__(self) -> None: + super().__init__( + skipkeys=False, + ensure_ascii=True, + check_circular=True, + allow_nan=False, + sort_keys=False, + indent=None, + separators=None, + default=None, + ) + + @singledispatchmethod + def default(self, o): + return super().default(o) + + @default.register + def _(self, o: Decimal) -> float: + return round(float(o), 4) + + @default.register + def _(self, o: datetime) -> str: + return o.astimezone(UTC).strftime(self.date_fmt) + + @default.register + def _(self, o: date) -> str: + return o.strftime(self.date_fmt) + + @default.register + def _(self, o: bytes) -> str: + return o.decode("utf-8") + + @default.register + def _(self, o: UUID) -> str: + return str(o) + + @default.register + def _(self, o: DetailAddenda) -> str: + return b64encode(self.encode(o.to_dict()).encode()).decode() + + @default.register + def _(self, o: SplitRule) -> dict: + return asdict(o) + + @default.register + def _(self, o: FiscalItem) -> dict: + return asdict(o) + + +def is_sandbox(key: str, /) -> bool: + return key.startswith("sandbox_") + + +def post( + endpoint: Endpoint, + /, + data: AnyStr, + signature: AnyStr, + *, + session: "Session", + stream: bool = False, + allow_redirects: bool = False, + proxies: Optional["Proxies"] = None, + timeout: Optional["Timeout"] = None, + hooks: Optional["Hooks"] = None, + verify: Optional["Verify"] = None, + cert: Optional["Cert"] = None, +) -> "Response": + """ + Send POST request to LiqPay API. + + Arguments + --------- + - `endpoint` -- API endpoint to send request to (see `liqpy.Endpoint`) + - `data` -- base64 encoded JSON data to send + - `signature` -- LiqPay signature for the data + - `session` -- `requests.Session` instance to use + - `stream` -- whether to stream the response + - `allow_redirects` -- whether to follow redirects + - `proxies` -- proxies to use + (see [Requests Proxies](https://docs.python-requests.org/en/stable/user/advanced/#proxies)) + - `timeout` -- timeout for the request + - `hooks` -- hooks for the request + (see [Requests Event Hooks](https://docs.python-requests.org/en/stable/user/advanced/#event-hooks)) + - `verify` -- whether to verify SSL certificate + (see [Request SSL Cert Verification](https://requests.readthedocs.io/en/stable/user/advanced/#ssl-cert-verification)) + - `cert` -- client certificate to use + (see [Request Client Side Certificates](https://requests.readthedocs.io/en/stable/user/advanced/#client-side-certificates)) + + Returns + ------- + - `requests.Response` instance + + Example + ------- + >>> from requests import Session + >>> from liqpy.api import encode, sign, request, Endpoint + >>> data = encode({"action": "status", "version": 3}) + >>> signature = sign(data, key=b"a4825234f4bae72a0be04eafe9e8e2bada209255") + >>> with Session() as session: # doctest: +SKIP + ... response = request(Endpoint.REQUEST, data, signature, session=session) # doctest: +SKIP + ... result = response.json() # doctest: +SKIP + """ + response = session.request( + method="POST", + url=endpoint.url(), + data={"data": data, "signature": signature}, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + json=None, + params=None, + cookies=None, + files=None, + auth=None, + proxies=proxies, + timeout=timeout, + hooks=hooks, + allow_redirects=allow_redirects, + stream=stream, + verify=verify, + cert=cert, + ) + response.raise_for_status() + return response + + +def sign(data: bytes, /, key: bytes) -> bytes: + """ + Sign data string with private key. + + >>> data = encode({"action": "status", "version": 3}) + >>> sign(data, key=b"a4825234f4bae72a0be04eafe9e8e2bada209255") + b'qI0/snsDFB7MiYUxrqhBqX2420E=' + """ + return b64encode(sha1(key + data + key).digest()) + + +def encode( + params: "LiqpayRequestDict", + /, + *, + filter_none: bool = True, + validator: Optional[BaseValidator] = None, + encoder: Optional[JSONEncoder] = None, + preprocessor: Optional[BasePreprocessor] = None, +) -> bytes: + """ + Encode parameters into base64 encoded JSON. + + >>> encode({"action": "status", "version": 3}) + b'eyJhY3Rpb24iOiAic3RhdHVzIiwgInZlcnNpb24iOiAzfQ==' + """ + if filter_none: + params = {key: value for key, value in params.items() if value is not None} + + if encoder is None: + encoder = LiqPayJSONEncoder() + + if preprocessor is None: + preprocessor = Preprocessor() + + preprocessor(params, encoder=encoder) + + if validator is None: + validator = Validator() + + validator(params) + + return b64encode(encoder.encode(params).encode()) + + +def decode(data: bytes, /) -> dict[str, Any]: + """Decode base64 encoded JSON.""" + return loads(b64decode(data)) + + +def request( + action: "Action", + /, + public_key: str, + *, + version: int = VERSION, + **params: "Unpack[LiqpayRequestDict]", +) -> "LiqpayRequestDict": + """ + Create data dictionary for LiqPay API request. + + >>> request("status", key="...", order_id="a1a1a1a1") + {'action': 'status', 'public_key': '...', 'version': 3, 'order_id': 'a1a1a1a1'} + """ + params.update(action=action, public_key=public_key, version=version) + + match action: + case "subscribe": + subscribe_date_start = params.get("subscribe_date_start") + + if subscribe_date_start is None: + subscribe_date_start = datetime.now(UTC) + + assert "subscribe_periodicity" in params, "subscribe_periodicity is required" + + params.update( + subscribe=True, + subscribe_date_start=subscribe_date_start, + ) + + case "letter_of_credit": + params["letter_of_credit"] = True + + return params diff --git a/liqpy/client.py b/liqpy/client.py index 1802301..dbed7e6 100644 --- a/liqpy/client.py +++ b/liqpy/client.py @@ -1,39 +1,29 @@ -from typing import Optional, Literal, Union, TYPE_CHECKING, Iterable -from hashlib import sha1 -from base64 import b64encode, b64decode +from typing import Optional, Literal, Union, TYPE_CHECKING, Unpack, AnyStr from os import environ -from json import dumps, loads from logging import getLogger -from datetime import datetime +from datetime import datetime, timedelta from numbers import Number from re import search +from uuid import UUID -from requests import Session, Response +from requests import Session from secret_type import secret, Secret -from .constants import VERSION, REQUEST_URL, CHECKOUT_URL -from .exceptions import exception_factory, is_exception, LiqPayException -from .util import ( - to_milliseconds, - to_dict, - is_sandbox, - format_date, - filter_none, - verify_url, -) +from liqpy import __version__ + +from .api import post, Endpoint, sign, request, encode, decode, VERSION, is_sandbox +from .exceptions import exception_factory +from .data import LiqpayCallback if TYPE_CHECKING: - from .types import ( - CallbackDict, - Language, - Format, - Currency, - PayType, - RROInfoDict, - SplitRuleDict, - SubscribePeriodicity, - DetailAddendaDict, - ) + from json import JSONEncoder + + from .preprocess import BasePreprocessor + from .validation import BaseValidator + + from .types.common import Language, Currency, SubscribePeriodicity, PayOption + from .types.request import Format, Language, LiqpayRequestDict + from .types.callback import LiqpayCallbackDict __all__ = ["Client"] @@ -42,33 +32,44 @@ logger = getLogger(__name__) +CHECKOUT_ACTIONS = ( + "auth", + "pay", + "hold", + "subscribe", + "paydonate", +) + + class Client: """ [LiqPay API](https://www.liqpay.ua/en/documentation/api/home) authorized client. - Intialize by setting environment variables - `LIQPAY_PUBLIC_KEY` and `LIQPAY_PRIVATE_KEY`: - >>> from liqpy.client import Client - >>> client = Client() + Intialize by setting environment variables `LIQPAY_PUBLIC_KEY` and `LIQPAY_PRIVATE_KEY`: + >>> client = Client() # doctest: +SKIP Or pass them as arguments: - >>> client = Client(public_key="...", private_key="...") + >>> Client(public_key="i00000000", private_key="a4825234f4bae72a0be04eafe9e8e2bada209255") + Client(public_key="i00000000") - For using custom session object pass it as an keyword argument: - >>> from requests import Session - >>> with Session() as session: - >>> client = Client(session=session) + For using custom [session](https://requests.readthedocs.io/en/stable/api/#request-sessions) + pass it as an keyword argument: + >>> with Session() as session: # doctest: +SKIP + >>> client = Client(session=session) # doctest: +SKIP - Client implements context manager interface same as `requests.Session`: - >>> with Client() as client: - >>> pass + Client implements context manager interface: + >>> with Client() as client: # doctest: +SKIP + >>> pass # doctest: +SKIP >>> # client.session is closed """ - session: Session - + _session: Session _public_key: str - _private_key: Secret[str] + _private_key: Secret[bytes] + + validator: Optional["BaseValidator"] = None + preprocessor: Optional["BasePreprocessor"] = None + encoder: Optional["JSONEncoder"] = None def __init__( self, @@ -77,9 +78,16 @@ def __init__( private_key: str | None = None, *, session: Session = None, + validator: Optional["BaseValidator"] = None, + preprocessor: Optional["BasePreprocessor"] = None, + encoder: Optional["JSONEncoder"] = None, ): - self.update_keys(public_key, private_key) - self.session = Session() if session is None else session + self.update_keys(public_key=public_key, private_key=private_key) + self.session = session + + self.validator = validator + self.preprocessor = preprocessor + self.encoder = encoder @property def public_key(self) -> str: @@ -91,24 +99,38 @@ def sandbox(self) -> bool: """Check if client use sandbox LiqPay API.""" return is_sandbox(self._public_key) - def update_keys(self, /, public_key: str | None, private_key: str | None) -> None: + @property + def session(self) -> Session: + return self._session + + @session.setter + def session(self, /, session: Optional[Session]): + if session is None: + session = Session() + else: + assert isinstance( + session, Session + ), "Session must be an instance of `requests.Session`" + + session.headers.update({"User-Agent": f"{__package__}/{__version__}"}) + self._session = session + + def update_keys( + self, /, *, public_key: str | None, private_key: str | None + ) -> None: """Update public and private keys.""" if public_key is None: public_key = environ["LIQPAY_PUBLIC_KEY"] - else: - public_key = str(public_key) if private_key is None: private_key = environ["LIQPAY_PRIVATE_KEY"] - else: - private_key = str(private_key) sandbox = is_sandbox(public_key) if sandbox != is_sandbox(private_key): raise ValueError("Public and private keys must be both sandbox or both not") self._public_key = public_key - self._private_key = secret(private_key) + self._private_key = secret(private_key.encode()) if sandbox: logger.warning("Using sandbox LiqPay API.") @@ -120,284 +142,162 @@ def __enter__(self): return self def __exit__(self, *args): - self.session.close() + self._session.close() def __del__(self): - self.session.close() - - def _prepare(self, /, action: str, **kwargs) -> dict: - return { - **filter_none(kwargs), - "public_key": self._public_key, - "version": VERSION, - "action": str(action), - } - - def _post(self, url: str, /, data: str, signature: str, **kwargs) -> "Response": - response = self.session.post(url, data=to_dict(data, signature), **kwargs) - response.raise_for_status() - return response - - def _post_request( - self, - /, - data: str, - signature: str, - *, - stream: bool = False, - **kwargs, - ) -> "Response": - return self._post(REQUEST_URL, data, signature, stream=stream, **kwargs) - - def _post_checkout( - self, - /, - data: str, - signature: str, - *, - redirect: bool = False, - **kwargs, - ) -> "Response": - return self._post( - CHECKOUT_URL, data, signature, allow_redirects=redirect, **kwargs - ) + self._session.close() def _callback( - self, data: str, signature: str, *, verify: bool = True - ) -> "CallbackDict": + self, /, data: bytes, signature: bytes, *, verify: bool = True + ) -> "LiqpayCallbackDict": if verify: self.verify(data, signature) else: - logger.warning("Skipping signature verification") + logger.warning("Skipping LiqPay signature verification") - return loads(b64decode(data)) + return decode(data) - def sign(self, data: str, /) -> str: - """Sign data string with private key.""" + def sign(self, data: bytes, /) -> bytes: + """ + Sign data string with private key. + + See `liqpy.api.sign` for more information. + """ with self._private_key.dangerous_reveal() as pk: - payload = f"{pk}{data}{pk}".encode() - return b64encode(sha1(payload).digest()).decode() + return sign(data, key=pk) - def encode(self, /, action: str, **kwargs) -> tuple[str, str]: + def encode( + self, /, action: str, **kwargs: Unpack["LiqpayRequestDict"] + ) -> tuple[bytes, bytes]: """ Encode parameters into data and signature strings. - See usage example in `liqpy.Client.callback`. + + >>> data, signature = client.encode("status", order_id="a1a1a1a1") + + See `liqpy.api.encode` for more information. """ - data = dumps(self._prepare(action, **kwargs)) - data = b64encode(data.encode()).decode() + data = encode( + request(action, public_key=self._public_key, version=VERSION, **kwargs), + filter_none=True, + validator=self.validator, + preprocessor=self.preprocessor, + encoder=self.encoder, + ) signature = self.sign(data) return data, signature - def is_valid(self, /, data: str, signature: str) -> bool: + def is_valid(self, /, data: bytes, signature: bytes) -> bool: """ Check if the signature is valid. Used for verification in `liqpy.Client.verify`. """ return self.sign(data) == signature - def verify(self, /, data: str, signature: str) -> None: + def verify(self, /, data: bytes, signature: bytes) -> None: """ Verify if the signature is valid. Raise an `AssertionError` if not. Used for verification in `liqpy.Client.callback`. """ assert self.is_valid(data, signature), "Invalid signature" - def request(self, action: str, **kwargs) -> dict: + def request(self, action: str, **kwargs: "LiqpayRequestDict") -> dict: """ Make a Server-Server request to LiqPay API. """ - response = self._post_request(*self.encode(action, **kwargs)) + response = post( + Endpoint.REQUEST, + *self.encode(action, **kwargs), + session=self._session, + allow_redirects=False, + stream=False, + ) if not response.headers.get("Content-Type", "").startswith("application/json"): - raise LiqPayException(response=response) + raise exception_factory(response=response) + + data: dict = response.json() + + result: Optional[Literal["ok", "error"]] = data.pop("result", None) + status = data.get("status") + err_code = data.pop("err_code", data.pop("code", None)) - result: dict = response.json() + if result == "ok" or (action in ("status", "data") and err_code is None): + return data - if is_exception(action, result.pop("result", ""), result.get("status")): + if status in ("error", "failure") or result == "error": raise exception_factory( - code=result.pop("err_code", None), - description=result.pop("err_description", None), + code=err_code, + description=data.pop("err_description", None), response=response, - details=result, + details=data, ) - return result + return data - def checkout( + def pay( self, /, + amount: Number, + order_id: str | UUID, + card: str, + currency: "Currency", + description: str, + **kwargs: "LiqpayRequestDict", + ) -> dict: + return self.request( + "pay", + order_id=order_id, + amount=amount, + card=card, + currency=currency, + description=description, + **kwargs, + ) + + def unsubscribe(self, /, order_id: str | UUID) -> dict: + return self.request("unsubscribe", order_id=order_id) + + def refund(self, /, order_id: str | UUID, amount: Number) -> dict: + return self.request("refund", order_id=order_id, amount=amount) + + def checkout( + self, action: Literal["auth", "pay", "hold", "subscribe", "paydonate"], - order_id: str, - *, + /, + order_id: str | UUID, amount: Number, currency: "Currency", description: str, - rro_info: Optional["RROInfoDict"] = None, - expired_date: Optional[Union[datetime, str, Number]] = None, - language: Optional["Language"] = None, - paytypes: Optional[Iterable["PayType"]] = None, - result_url: Optional[str] = None, - server_url: Optional[str] = None, - verifycode: bool = False, - split_rules: Optional[Iterable["SplitRuleDict"]] = None, - sender_address: Optional[str] = None, - sender_city: Optional[str] = None, - sender_country_code: Optional[str] = None, - sender_first_name: Optional[str] = None, - sender_last_name: Optional[str] = None, - sender_postal_code: Optional[str] = None, - letter_of_credit: Optional[str] = None, - letter_of_credit_date: Optional[Union[datetime, str, Number]] = None, - subscribe_date_start: Optional[Union[datetime, str, Number]] = None, - subscribe_periodicity: Optional["SubscribePeriodicity"] = None, - customer: Optional[str] = None, - recurring_by_token: bool = False, - customer_user_id: Optional[str] = None, - detail_addenda: Optional["DetailAddendaDict"] = None, - info: Optional[str] = None, - product_category: Optional[str] = None, - product_description: Optional[str] = None, - product_name: Optional[str] = None, - product_url: Optional[str] = None, - **kwargs, + expired_date: str | datetime | None = None, + paytypes: Optional[list["PayOption"]] = None, + **kwargs: Unpack["LiqpayRequestDict"], ) -> str: """ Make a Client-Server checkout request to LiqPay API. - `kwargs` are passed to `requests.Session.post` method. + Returns a url to redirect the user to. [Documentation](https://www.liqpay.ua/en/documentation/api/aquiring/checkout/doc) """ - assert action in ( - "auth", - "pay", - "hold", - "subscribe", - "paydonate", - ), "Invalid action. Must be one of: auth, pay, hold, subscribe, paydonate" - - assert isinstance(amount, Number), "Amount must be a number" - - assert currency in ( - "EUR", - "UAH", - "USD", - ), "Invalid currency. Must be one of: EUR, UAH, USD" - - order_id = str(order_id) - assert len(order_id) <= 255, "Order id must be less than 255 characters" - - params = { - "order_id": order_id, - "amount": amount, - "currency": currency, - "description": str(description), - "rro_info": rro_info, - "sender_address": sender_address, - "sender_city": sender_city, - "sender_country_code": sender_country_code, - "sender_first_name": sender_first_name, - "sender_last_name": sender_last_name, - "sender_postal_code": sender_postal_code, - "letter_of_credit": letter_of_credit, - "customer_user_id": customer_user_id, - "info": info, - } - - if action == "auth" and verifycode: - params["verifycode"] = "Y" - - if language is not None: - assert language in ("en", "uk"), "Invalid language. Must be one of: en, uk" - params["language"] = language - - if result_url is not None: - verify_url(result_url) - params["result_url"] = result_url - - if server_url is not None: - verify_url(server_url) - params["server_url"] = server_url - - if paytypes is not None: - paytypes = set(paytypes) - assert paytypes.issubset( - ( - "card", - "liqpay", - "privat24", - "masterpass", - "moment_part", - "cash", - "invoice", - "qr", - ) - ), "Invalid paytypes. Must be one of: card, liqpay, privat24, masterpass, moment_part, cash, invoice, qr" - params["paytypes"] = ",".join(paytypes) - - if action == "subscribe": - if subscribe_date_start is None: - subscribe_date_start = datetime.utcnow() - - if subscribe_periodicity is not None: - assert subscribe_periodicity in ( - "month", - "year", - ), "Invalid subscribe periodicity. Must be one of: month, year" - - params.update( - subscribe=1, - subscribe_date_start=format_date(subscribe_date_start), - subscribe_periodicity=subscribe_periodicity or "month", - ) - - if customer is not None: - assert len(customer) <= 100, "Customer must be less than 100 characters" - params["customer"] = customer - - if expired_date is not None: - params["expired_date"] = format_date(expired_date) - - if split_rules is not None and len(split_rules) > 0: - params["split_rules"] = dumps(list(split_rules)) - - if letter_of_credit_date is not None: - params["letter_of_credit_date"] = format_date(letter_of_credit_date) - - if recurring_by_token: - assert ( - server_url is not None - ), "Server url must be specified for recurring by token" - params["recurringbytoken"] = "1" - - if detail_addenda is not None: - params["dae"] = b64encode(dumps(detail_addenda).encode()).decode() - - if product_category is not None: - assert ( - len(product_category) <= 25 - ), "Product category must be less than 25 characters" - params["product_category"] = product_category - - if product_description is not None: - assert ( - len(product_description) <= 500 - ), "Product description must be less than 500 characters" - params["product_description"] = product_description - - if product_name is not None: - assert ( - len(product_name) <= 100 - ), "Product name must be less than 100 characters" - params["product_name"] = product_name - - if product_url is not None: - verify_url(product_url) - params["product_url"] = product_url - - response = self._post_checkout( - *self.encode(action, **params), redirect=False, **kwargs + assert ( + action in CHECKOUT_ACTIONS + ), "Invalid action. Must be one of: %s" % ",".join(CHECKOUT_ACTIONS) + + response = post( + Endpoint.CHECKOUT, + *self.encode( + action, + order_id=order_id, + amount=amount, + currency=currency, + description=description, + expired_date=expired_date, + paytypes=paytypes, + **kwargs, + ), + session=self._session, + allow_redirects=False, ) next = response.next @@ -411,7 +311,7 @@ def checkout( code=result.pop("err_code", None), description=result.pop("err_description", None), response=response, - details=result, + details=result if len(result) else None, ) return next.url @@ -419,8 +319,8 @@ def checkout( def reports( self, /, - date_from: Union[datetime, str, Number], - date_to: Union[datetime, str, Number], + date_from: Union[datetime, str, int], + date_to: Union[datetime, str, int], *, format: Optional["Format"] = None, ) -> str: @@ -440,16 +340,13 @@ def reports( [Documentaion](https://www.liqpay.ua/en/documentation/api/information/reports/doc) """ - - kwargs = { - "date_from": to_milliseconds(date_from), - "date_to": to_milliseconds(date_to), - } - - if format is not None: - kwargs["resp_format"] = format - - response = self._post_request(*self.encode("reports", **kwargs)) + response = post( + Endpoint.REQUEST, + *self.encode( + "reports", date_from=date_from, date_to=date_to, resp_format=format + ), + session=self._session, + ) output: str = response.text error: dict | None = None @@ -477,6 +374,36 @@ def reports( details=error, ) + def subscribe( + self, + /, + order_id: str | UUID, + amount: Number, + card: str, + card_cvv: str, + card_exp_month: str, + card_exp_year: str, + currency: "Currency", + description: str, + subscribe_periodicity: "SubscribePeriodicity", + subscribe_date_start: datetime | str | timedelta | None | Number, + **kwargs: Unpack["LiqpayRequestDict"], + ) -> dict: + return self.request( + "subscribe", + order_id=order_id, + amount=amount, + card=card, + card_cvv=card_cvv, + card_exp_month=card_exp_month, + card_exp_year=card_exp_year, + currency=currency, + description=description, + subscribe_date_start=subscribe_date_start, + subscribe_periodicity=subscribe_periodicity, + **kwargs, + ) + def data(self, /, order_id: str, info: str) -> dict: """ Adding an info to already created payment. @@ -499,16 +426,15 @@ def receipt( [Documentation](https://www.liqpay.ua/en/documentation/api/information/ticket/doc) """ - kwargs = {} - if payment_id is not None: - kwargs["payment_id"] = payment_id - - if language is not None: - kwargs["language"] = language - - self.request("receipt", order_id=order_id, email=email, **kwargs) + self.request( + "receipt", + order_id=order_id, + email=email, + payment_id=payment_id, + language=language, + ) - def status(self, order_id: str, /) -> dict: + def status(self, order_id: str | UUID, /) -> dict: """ Get the status of a payment. @@ -516,23 +442,21 @@ def status(self, order_id: str, /) -> dict: """ return self.request("status", order_id=order_id) - def callback(self, /, data: str, signature: str, *, verify: bool = True): + def callback(self, /, data: AnyStr, signature: AnyStr, *, verify: bool = True): """ Verify and decode the callback data. Example: - >>> from uuid import uuid4 - >>> from liqpy.client import Client >>> client = Client() >>> # get data and signature from webhook request body - >>> order_id = str(uuid4()) + >>> order_id = "a1a1a1a1a1" >>> data, signature = client.encode( - >>> action="pay", - >>> amount=1, - >>> order_id=order_id, - >>> description="Test Encoding", - >>> currency="USD", - >>> ) + ... action="pay", + ... amount=1, + ... order_id=order_id, + ... description="Test Encoding", + ... currency="USD", + ... ) >>> # verify and decode data >>> result = client.callback(data, signature) >>> assert result["order_id"] == order_id @@ -543,10 +467,19 @@ def callback(self, /, data: str, signature: str, *, verify: bool = True): [Documentation](https://www.liqpay.ua/en/documentation/api/callback) """ + if isinstance(data, str): + data = data.encode() + + if isinstance(signature, str): + signature = signature.encode() + result = self._callback(data, signature, verify=verify) version = result.get("version") if version != VERSION: logger.warning("Callback version mismatch: %s != %s", version, VERSION) - return result + try: + return LiqpayCallback(**result) + finally: + logger.warning("Failed to parse callback data.", extra=result) diff --git a/liqpy/constants.py b/liqpy/constants.py deleted file mode 100644 index dec2de0..0000000 --- a/liqpy/constants.py +++ /dev/null @@ -1,5 +0,0 @@ -VERSION = 3 -BASE_URL = "https://www.liqpay.ua" -API_URL = f"{BASE_URL}/api" -REQUEST_URL = f"{API_URL}/request" -CHECKOUT_URL = f"{API_URL}/{VERSION}/checkout" diff --git a/liqpy/convert.py b/liqpy/convert.py new file mode 100644 index 0000000..7f28a9b --- /dev/null +++ b/liqpy/convert.py @@ -0,0 +1,98 @@ +from typing import overload, TYPE_CHECKING +from functools import singledispatch, cache +from numbers import Number +from datetime import datetime, UTC, timedelta +from re import compile + + +@cache +def date_pattern(flags: int = 0): + return compile(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", flags=flags) + + +@singledispatch +def to_datetime(value, **kwargs) -> datetime: + raise NotImplementedError(f"Unsupported type: {type(value)}") + + +@to_datetime.register +def _(value: datetime, **kwargs): + return value + + +@to_datetime.register +def _(value: str, **kwargs): + if date_pattern().fullmatch(value): + return datetime.strptime(value, r"%Y-%m-%d %H:%M:%S") + else: + return datetime.fromisoformat(value) + + +@to_datetime.register +def _(value: Number, tz=UTC, **kwargs): + return datetime.fromtimestamp(float(value), tz=tz) + + +@to_datetime.register +def _(value: timedelta, tz=UTC, **kwargs): + return datetime.now(tz) + value + + +@singledispatch +def to_milliseconds(value, **kwargs) -> int: + raise NotImplementedError(f"Unsupported type: {type(value)}") + + +@to_milliseconds.register +def _(value: int, **kwargs): + return value + + +@to_milliseconds.register +def _(value: datetime, **kwargs): + return int(value.timestamp() * 1000) + + +@to_milliseconds.register +def _(value: str, **kwargs): + return to_milliseconds(to_datetime(value, **kwargs)) + + +@to_milliseconds.register +def _(value: timedelta, **kwargs): + return to_milliseconds(to_datetime(value, **kwargs)) + + +if TYPE_CHECKING: + + @overload + def to_datetime(value: Number, **kwargs) -> datetime: + ... + + @overload + def to_datetime(value: datetime, **kwargs) -> datetime: + ... + + @overload + def to_datetime(value: str, **kwargs) -> datetime: + ... + + @overload + def to_datetime(value: timedelta, **kwargs) -> datetime: + ... + + @overload + def to_milliseconds(value: datetime, **kwargs) -> int: + ... + + @overload + def to_milliseconds(value: str, **kwargs) -> int: + ... + + @overload + def to_milliseconds(value: int, **kwargs) -> int: + ... + + @overload + def to_milliseconds(value: timedelta, **kwargs) -> int: + ... diff --git a/liqpy/data.py b/liqpy/data.py new file mode 100644 index 0000000..184304c --- /dev/null +++ b/liqpy/data.py @@ -0,0 +1,150 @@ +from typing import Literal, TYPE_CHECKING, Optional +from dataclasses import dataclass, asdict +from datetime import datetime +from numbers import Number +from ipaddress import ip_address, IPv4Address + +from .convert import to_datetime + +if TYPE_CHECKING: + from .types.common import Currency, Language, PayType + from .types.callback import ThreeDS, CallbackAction + from .types.error import LiqPayErrcode + from .types import status + + +def from_milliseconds(value: int) -> datetime: + return datetime.fromtimestamp(value / 1000) + + +@dataclass(kw_only=True) +class DetailAddenda: + air_line: str + ticket_number: str + passenger_name: str + flight_number: str + origin_city: str + destination_city: str + departure_date: datetime + + def __post_init__(self): + if not isinstance(self.departure_date, datetime): + self.departure_date = to_datetime(self.departure_date) + + def to_dict(self): + return { + "airLine": self.air_line, + "ticketNumber": self.ticket_number, + "passengerName": self.passenger_name, + "flightNumber": self.flight_number, + "originCity": self.origin_city, + "destinationCity": self.destination_city, + "departureDate": self.departure_date.strftime(r"%d%m%y"), + } + + +@dataclass(kw_only=True) +class SplitRule: + public_key: str + amount: Number + commission_payer: Literal["sender", "receiver"] + server_url: str + + +@dataclass(kw_only=True) +class FiscalItem: + id: int + amount: Number + cost: Number + price: Number + + +@dataclass(kw_only=True) +class FiscalInfo: + items: list[FiscalItem] + delivery_emails: list[str] + + +@dataclass(init=False) +class LiqpayCallback: + acq_id: int + action: "CallbackAction" + agent_commission: Number + amount: Number + amount_bonus: Number + amount_credit: Number + amount_debit: Number + authcode_credit: str | None = None + authcode_debit: str | None = None + card_token: str | None = None + commission_credit: Number + commission_debit: Number + completion_date: datetime | None = None + create_date: datetime + currency: "Currency" + currency_credit: "Currency" + currency_debit: "Currency" + customer: str | None = None + description: str + end_date: datetime + err_code: Optional["LiqPayErrcode"] = None + err_description: str | None = None + info: str | None = None + ip: IPv4Address | None = None + is_3ds: bool + language: "Language" + liqpay_order_id: str + mpi_eci: "ThreeDS" + order_id: str + payment_id: int + paytype: "PayType" + public_key: str + receiver_commission: Number + redirect_to: str | None = None + refund_date_last: datetime | None = None + rrn_credit: str | None = None + rrn_debit: str | None = None + sender_bonus: Number + sender_card_bank: str + sender_card_country: int + sender_card_mask2: str + sender_card_type: str + sender_commission: Number + sender_first_name: str | None = None + sender_last_name: str | None = None + sender_phone: str | None = None + status: "status.CallbackStatus" + transaction_id: str | None = None + token: str | None = None + type: str + version: Literal[3] + err_erc: Optional["LiqPayErrcode"] = None + product_category: str | None = None + product_description: str | None = None + product_name: str | None = None + product_url: str | None = None + refund_amount: Number | None = None + verifycode: str | None = None + + code: str | None = None + + def __init__(self, **kwargs): + for k, v in kwargs.items(): + setattr(self, k, v) + + self.__post_init__() + + def __post_init__(self): + self.create_date = from_milliseconds(self.create_date) + self.end_date = from_milliseconds(self.end_date) + + if self.completion_date is not None: + self.completion_date = from_milliseconds(self.completion_date) + + if self.refund_date_last is not None: + self.refund_date_last = from_milliseconds(self.refund_date_last) + + if self.ip is not None: + self.ip = ip_address(self.ip) + + self.mpi_eci = int(self.mpi_eci) diff --git a/liqpy/exceptions.py b/liqpy/exceptions.py index c745f3c..fb19a4e 100644 --- a/liqpy/exceptions.py +++ b/liqpy/exceptions.py @@ -1,27 +1,29 @@ -from typing import Literal, TYPE_CHECKING, Optional, Union -from typing import get_args +from typing import TYPE_CHECKING, Optional if TYPE_CHECKING: from requests import Response + from .types.error import ( + LiqPayErrcode, + LiqpayAntiFraudErrcode, + LiqpayFinancialErrcode, + LiqpayNonFinancialErrcode, + LiqpayExpireErrcode, + LiqpayRequestErrcode, + LiqpayPaymentErrcode, + ) + UNKNOWN_ERRCODE = "unknown" UNKNOWN_ERRMSG = "Unknown error" - -def is_exception( - action: str, - result: Literal["error", "ok"], - status: Literal["error", "failure", "success"], -) -> bool: - if result == "error" or status in ["error", "failure"]: - if action != "status" or status == "error": - return True - return False +TRANSLATIONS = { + "Платеж не найден": "Payment not found", +} class LiqPayException(Exception): - code: str + code: "LiqPayErrcode" details: dict response: Optional["Response"] = None @@ -34,157 +36,63 @@ def __init__( response: Optional["Response"] = None, details: Optional[dict] = None, ): + description = TRANSLATIONS.get(description, description) + super().__init__(description or UNKNOWN_ERRMSG) self.code = code or UNKNOWN_ERRCODE self.response = response self.details = details -LiqpayAntiFraudErrcode = Literal["limit", "frod", "decline"] - - -LiqpayNonFinancialErrcode = Literal[ - "err_auth", - "err_cache", - "user_not_found", - "err_sms_send", - "err_sms_otp", - "shop_blocked", - "shop_not_active", - "invalid_signature", - "order_id_empty", - "err_shop_not_agent", - "err_card_def_notfound", - "err_no_card_token", - "err_card_liqpay_def", - "err_card_type", - "err_card_country", - "err_limit_amount", - "err_payment_amount_limit", - "amount_limit", - "payment_err_sender_card", - "payment_processing", - "err_payment_discount", - "err_wallet", - "err_get_verify_code", - "err_verify_code", - "wait_info", - "err_path", - "err_payment_cash_acq", - "err_split_amount", - "err_card_receiver_def", - "payment_err_status", - "public_key_not_found", - "payment_not_found", - "payment_not_subscribed", - "wrong_amount_currency", - "err_amount_hold", - "err_access", - "order_id_duplicate", - "err_blocked", - "err_empty", - "err_empty_phone", - "err_missing", - "err_wrong", - "err_wrong_currency", - "err_phone", - "err_card", - "err_card_bin", - "err_terminal_notfound", - "err_commission_notfound", - "err_payment_create", - "err_mpi", - "err_currency_is_not_allowed", - "err_look", - "err_mods_empty", - "payment_err_type", - "err_payment_currency", - "err_payment_exchangerates", - "err_signature", - "err_api_action", - "err_api_callback", - "err_api_ip", - "expired_phone", - "expired_3ds", - "expired_otp", - "expired_cvv", - "expired_p24", - "expired_sender", - "expired_pin", - "expired_ivr", - "expired_captcha", - "expired_password", - "expired_senderapp", - "expired_prepared", - "expired_mp", - "expired_qr", - "5", -] - -LiqpayFinancialErrcode = Literal[ - "90", - "101", - "102", - "103", - "104", - "105", - "106", - "107", - "108", - "109", - "110", - "111", - "112", - "113", - "114", - "115", - "2903", - "2915", - "3914", - "9851", - "9852", - "9854", - "9855", - "9857", - "9859", - "9860", - "9861", - "9863", - "9867", - "9868", - "9872", - "9882", - "9886", - "9961", - "9989", -] - -Errcode = Union[ - LiqpayAntiFraudErrcode, - LiqpayFinancialErrcode, - LiqpayNonFinancialErrcode, -] - - class LiqPayAntiFraudException(LiqPayException): - code: LiqpayAntiFraudErrcode + code: "LiqpayAntiFraudErrcode" class LiqPayNonFinancialException(LiqPayException): - code: LiqpayNonFinancialErrcode + code: "LiqpayNonFinancialErrcode" + + +class LiqPayExpireException(LiqPayNonFinancialException): + code: "LiqpayExpireErrcode" + + +class LiqPayRequestException(LiqPayNonFinancialException): + code: "LiqpayRequestErrcode" + + +class LiqPayPaymentException(LiqPayNonFinancialException): + code: "LiqpayPaymentErrcode" class LiqPayFinancialException(LiqPayException): - code: LiqpayFinancialErrcode + code: "LiqpayFinancialErrcode" def get_exception_cls(code: str | None = None) -> type[LiqPayException]: - if code in get_args(LiqpayAntiFraudErrcode): + if code is None or code == "unknown": + return LiqPayException + elif code in ("limit", "frod", "decline"): return LiqPayAntiFraudException - elif code in get_args(LiqpayFinancialErrcode): + elif code.isdigit() and code != "5": return LiqPayFinancialException - elif code in get_args(LiqpayNonFinancialErrcode): + elif code.startswith("expired_"): + return LiqPayExpireException + elif code.startswith("err_"): return LiqPayNonFinancialException + elif code.startswith(("shop_")): + return LiqPayNonFinancialException + elif code.endswith(("_not_found", "_limit")): + return LiqPayNonFinancialException + elif code.startswith("payment_"): + return LiqPayExpireException + elif code in ( + "invalid_signature", + "public_key_not_found", + "order_id_empty", + "amount_limit", + "wrong_amount_currency", + ): + return LiqPayRequestException else: return LiqPayException diff --git a/liqpy/preprocess.py b/liqpy/preprocess.py new file mode 100644 index 0000000..3f12fcc --- /dev/null +++ b/liqpy/preprocess.py @@ -0,0 +1,79 @@ +from typing import TYPE_CHECKING, Optional +from datetime import timedelta + +from .convert import to_datetime, to_milliseconds +from .data import DetailAddenda + +if TYPE_CHECKING: + from json import JSONEncoder + from .types.request import LiqpayRequestDict + + +class BasePreprocessor: + def __call__( + self, o: "LiqpayRequestDict", /, encoder: Optional["JSONEncoder"], **kwargs + ): + if encoder is None: + encoder = JSONEncoder() + + for key, value in o.items(): + try: + fn = getattr(self, key, None) + + if not callable(fn): + continue + + processed = fn(value, encoder=encoder, **kwargs.get(key, {})) + + if processed is not None: + o[key] = processed + + except Exception as e: + raise Exception(f"Failed to convert {key} parameter.") from e + + +class Preprocessor(BasePreprocessor): + def dae(self, value, /, **kwargs): + if isinstance(value, DetailAddenda): + return value + else: + return DetailAddenda(**value) + + def split_rules(self, value, /, encoder: "JSONEncoder", **kwargs): + if isinstance(value, list): + return encoder.encode(value) + + def paytypes(self, value, /, **kwargs): + if isinstance(value, list): + return ",".join(value) + + def date_from(self, value, /, **kwargs): + return to_milliseconds(value, **kwargs) + + def date_to(self, value, /, **kwargs): + return to_milliseconds(value, **kwargs) + + def subscribe_date_start(self, value, /, **kwargs): + return to_datetime(value, **kwargs) + + def letter_of_credit_date(self, value, /, **kwargs): + return to_datetime(value, **kwargs) + + def expired_date(self, value, /, **kwargs): + return to_datetime(value, **kwargs) + + def verifycode(self, value, /, **kwargs): + if value: + return "Y" + + def subscribe(self, value, /, **kwargs): + if value: + return 1 + + def letter_of_credit(self, value, /, **kwargs): + if value: + return 1 + + def recurringbytoken(self, value, /, **kwargs): + if value: + return "1" diff --git a/tests/server.py b/liqpy/server.py similarity index 84% rename from tests/server.py rename to liqpy/server.py index ff86de1..75e9f32 100644 --- a/tests/server.py +++ b/liqpy/server.py @@ -3,10 +3,10 @@ from http.server import HTTPServer, BaseHTTPRequestHandler from urllib.parse import parse_qs -from liqpy.client import Client +from .client import Client if TYPE_CHECKING: - from liqpy.types import CallbackDict + from .types import LiqpayCallbackDict class LiqpayHandler(BaseHTTPRequestHandler): @@ -41,7 +41,7 @@ def _handle_webhook(self): data, signature = self._parse_body() return self.client.callback(data, signature, verify=True) - def _push_callback(self, callback: "CallbackDict"): + def _push_callback(self, callback: "LiqpayCallbackDict"): pprint(callback) self.server.callback_history.append(callback) @@ -58,23 +58,31 @@ def do_POST(self): class LiqpayServer(HTTPServer): client: "Client" - callback_history: List["CallbackDict"] + callback_history: List["LiqpayCallbackDict"] """Liqpay server for testing. Do not use in production!""" def __init__( self, - *, + /, host: str = "localhost", port: int = 8000, + *, client: Optional["Client"] = None, + timeout: float | None = None, ): super().__init__((host, port), LiqpayHandler) self.client = Client() if client is None else client self.callback_history = [] + if timeout is not None: + self.timeout = float(timeout) + + self.allow_reuse_address = True + self.allow_reuse_port = True + @property - def last_callback(self) -> "CallbackDict": + def last_callback(self) -> "LiqpayCallbackDict": return self.callback_history[-1] diff --git a/liqpy/testing.py b/liqpy/testing.py index 5e30934..a841648 100644 --- a/liqpy/testing.py +++ b/liqpy/testing.py @@ -7,9 +7,9 @@ __all__ = ["TestCard", "gen_card_expire", "gen_card_cvv", "fmt_card_expire_date"] -def fmt_card_expire_date(value: date): +def fmt_card_expire_date(value: date) -> tuple[str, str]: """Format a date object to MM, YY strings""" - return str(value.month).ljust(2, "0"), str(value.year)[-2:].ljust(2, "0") + return str(value.month).rjust(2, "0"), str(value.year)[-2:].rjust(2, "0") def gen_card_expire(valid: bool = True): @@ -20,13 +20,13 @@ def gen_card_expire(valid: bool = True): d += timedelta(days=randint(1, 365 * 4)) else: d -= timedelta(days=randint(1, 30 * 3)) - + return fmt_card_expire_date(d) def gen_card_cvv() -> str: """Generate a random CVV code""" - return str(randint(0, 999)).ljust(3, "0") + return str(randint(0, 999)).rjust(3, "0") class TestCard(Enum): diff --git a/liqpy/types.py b/liqpy/types.py deleted file mode 100644 index 27df2f7..0000000 --- a/liqpy/types.py +++ /dev/null @@ -1,160 +0,0 @@ -from typing import TypedDict, Literal, Union, TYPE_CHECKING, List - -if TYPE_CHECKING: - from numbers import Number - - from .exceptions import Errcode - - -class RequestForm(TypedDict): - data: str - signature: str - - -Format = Literal["json", "xml", "csv"] -Language = Literal["uk", "en"] -SubscribeAction = Literal["subscribe"] -Currency = Literal["UAH", "USD", "EUR"] - -WidgetAction = CheckoutAction = Literal[ - "pay", - "hold", - "paysplit", - "paydonate", - "auth", - "letter_of_credit", - "split_rules", - "apay", - "gpay", -] -CallbackAction = Union[ - SubscribeAction, Literal["pay", "hold", "paysplit", "paydonate", "auth", "regular"] -] -SubscriptionAction = Union[SubscribeAction, Literal["unsubscribe", "subscribe_update"]] - -SubscribeStatus = Literal["subscribed", "unsubscribed"] -FinalPaymentStatus = Union[ - Literal["error", "failure", "reversed", "success"], SubscribeStatus -] -ConfirmationStatus = Literal[ - "3ds_verify", - "captcha_verify", - "cvv_verify", - "ivr_verify", - "otp_verify", - "password_verify", - "phone_verify", - "pin_verify", - "receiver_verify", - "sender_verify", - "senderapp_verify", - "wait_qr", - "wait_sender", -] -OtherPaymentStatus = Literal[ - "cash_wait", - "hold_wait", - "invoice_wait", - "prepared", - "processing", - "wait_accept", - "wait_card", - "wait_compensation", - "wait_lc", - "wait_reserve", - "wait_secure", -] -CallbackStatus = Union[FinalPaymentStatus, ConfirmationStatus, OtherPaymentStatus] -MpiEci = Literal[5, 6, 7] -PayType = Literal[ - "card", "liqpay", "privat24", "masterpass", "moment_part", "cash", "invoice", "qr" -] -SubscribePeriodicity = Literal["month", "year"] - - -class DetailAddendaDict(TypedDict): - airLine: str - ticketNumber: str - passengerName: str - flightNumber: str - originCity: str - destinationCity: str - departureDate: str - - -class SplitRuleDict(TypedDict): - public_key: str - amount: "Number" - commission_payer: Literal["sender", "receiver"] - server_url: str - - -class ProductDict(TypedDict): - amount: "Number" - cost: "Number" - id: int - price: "Number" - - -class RROInfoDict(TypedDict, total=False): - items: List[ProductDict] - delivery_emails: List[str] - - -class CallbackDict(TypedDict, total=False): - acq_id: "Number" - action: CallbackAction - agent_commission: "Number" - amount: "Number" - amount_bonus: "Number" - amount_credit: "Number" - amount_debit: "Number" - authcode_credit: str - authcode_debit: str - card_token: str - commission_credit: "Number" - commission_debit: "Number" - completion_date: str - create_date: str - currency: Currency - currency_credit: str - currency_debit: str - customer: str - description: str - end_date: str - err_code: "Errcode" - err_description: str - info: str - ip: str - is_3ds: bool - liqpay_order_id: str - mpi_eci: MpiEci - order_id: str - payment_id: "Number" - paytype: PayType - public_key: str - receiver_commission: "Number" - redirect_to: str - refund_date_last: str - rrn_credit: str - rrn_debit: str - sender_bonus: "Number" - sender_card_bank: str - sender_card_country: str - sender_card_mask2: str - sender_card_type: str - sender_commission: "Number" - sender_first_name: str - sender_last_name: str - sender_phone: str - status: CallbackStatus - token: str - type: str - version: Literal[3] - err_erc: Errcode - product_category: str - product_description: str - product_name: str - product_url: str - refund_amount: "Number" - verifycode: str diff --git a/liqpy/types/__init__.py b/liqpy/types/__init__.py new file mode 100644 index 0000000..f52bdad --- /dev/null +++ b/liqpy/types/__init__.py @@ -0,0 +1,6 @@ +from . import common +from . import action +from . import status + +from .callback import LiqpayCallbackDict +from .request import LiqpayRequestDict diff --git a/liqpy/types/action.py b/liqpy/types/action.py new file mode 100644 index 0000000..f6e2645 --- /dev/null +++ b/liqpy/types/action.py @@ -0,0 +1,45 @@ +from typing import Literal as _Literal + + +Action = _Literal[ + "auth", + "pay", + "hold", + "payqr", + "paytoken", + "paycash", + "hold_completion", + "subscribe", + "paydonate", + "paysplit", + "split_rules", + "letter_of_credit", + "apay", + "gpay", + "regular", + "refund", + "payment_prepare", + "p2pcredit", + "p2pdebit", + "confirm", + "mpi", + "cardverification", + "register", + "data", + "ticket", + "status", + "invoice_send", + "invoice_cancel", + "token_create", + "token_create_unique", + "token_update", + "reports", + "reports_compensation", + "reports_compensation_file", + "agent_shop_create", + "agent_shop_edit", + "agent_info_merchant", + "agent_info_user", + "unsubscribe", + "subscribe_update", +] diff --git a/liqpy/types/callback.py b/liqpy/types/callback.py new file mode 100644 index 0000000..7e2ad7e --- /dev/null +++ b/liqpy/types/callback.py @@ -0,0 +1,73 @@ +from typing import Literal, TypedDict + +from numbers import Number + +from liqpy.exceptions import LiqPayErrcode +from liqpy.types import status +from liqpy.types.common import Currency, PayType, Language + + +CallbackAction = Literal[ + "pay", "hold", "paysplit", "subscribe", "paydonate", "auth", "regular" +] +ThreeDS = Literal[5, 6, 7] + + +class LiqpayCallbackDict(TypedDict, total=False): + acq_id: Number + action: CallbackAction + agent_commission: Number + amount: Number + amount_bonus: Number + amount_credit: Number + amount_debit: Number + authcode_credit: str + authcode_debit: str + card_token: str + commission_credit: Number + commission_debit: Number + completion_date: str + create_date: str + currency: Currency + currency_credit: str + currency_debit: str + customer: str + description: str + end_date: str + err_code: LiqPayErrcode + err_description: str + info: str + ip: str + is_3ds: bool + language: Language + liqpay_order_id: str + mpi_eci: ThreeDS | Literal["5", "6", "7"] + order_id: str + payment_id: int + paytype: PayType + public_key: str + receiver_commission: Number + redirect_to: str + refund_date_last: str + rrn_credit: str + rrn_debit: str + sender_bonus: Number + sender_card_bank: str + sender_card_country: str + sender_card_mask2: str + sender_card_type: str + sender_commission: Number + sender_first_name: str + sender_last_name: str + sender_phone: str + status: status.CallbackStatus + token: str + type: str + version: Literal[3] + err_erc: LiqPayErrcode + product_category: str + product_description: str + product_name: str + product_url: str + refund_amount: Number + verifycode: str diff --git a/liqpy/types/common.py b/liqpy/types/common.py new file mode 100644 index 0000000..fdfbfb5 --- /dev/null +++ b/liqpy/types/common.py @@ -0,0 +1,17 @@ +from typing import Literal as _Literal + +Language = _Literal["uk", "en"] +Currency = _Literal["UAH", "USD", "EUR"] +Format = _Literal["json", "xml", "csv"] + +PayType = _Literal[ + "apay", + "gpay", + "apay_tavv", + "gpay_tavv", + "tavv", +] +PayOption = _Literal[ + "card", "liqpay", "privat24", "masterpass", "moment_part", "cash", "invoice", "qr" +] +SubscribePeriodicity = _Literal["month", "year"] diff --git a/liqpy/types/error.py b/liqpy/types/error.py new file mode 100644 index 0000000..723f8d4 --- /dev/null +++ b/liqpy/types/error.py @@ -0,0 +1,143 @@ +from typing import Literal as _Literal, Union as _Union + + +UnknownErrcode = _Literal["unknown"] + +LiqpayAntiFraudErrcode = _Literal["limit", "frod", "decline"] + +LiqpayRequestErrcode = _Literal[ + "invalid_signature", + "public_key_not_found", + "order_id_empty", + "amount_limit", + "wrong_amount_currency", +] + +LiqpayExpireErrcode = _Literal[ + "expired_phone", + "expired_3ds", + "expired_otp", + "expired_cvv", + "expired_p24", + "expired_sender", + "expired_pin", + "expired_ivr", + "expired_captcha", + "expired_password", + "expired_senderapp", + "expired_prepared", + "expired_mp", + "expired_qr", +] + +LiqpayPaymentErrcode = _Literal[ + "payment_err_sender_card", + "payment_processing", + "payment_err_status", + "payment_not_found", + "payment_not_subscribed", + "payment_err_type", +] + +LiqpayNonFinancialErrcode = _Union[ + _Literal[ + "err_auth", + "err_cache", + "err_sms_send", + "err_sms_otp", + "err_shop_not_agent", + "err_card_def_notfound", + "err_no_card_token", + "err_card_liqpay_def", + "err_card_type", + "err_card_country", + "err_limit_amount", + "err_payment_amount_limit", + "err_payment_discount", + "err_wallet", + "err_get_verify_code", + "err_verify_code", + "err_path", + "err_payment_cash_acq", + "err_split_amount", + "err_card_receiver_def", + "err_amount_hold", + "err_access", + "err_blocked", + "err_empty", + "err_empty_phone", + "err_missing", + "err_wrong", + "err_wrong_currency", + "err_phone", + "err_card", + "err_card_bin", + "err_terminal_notfound", + "err_commission_notfound", + "err_payment_create", + "err_mpi", + "err_currency_is_not_allowed", + "err_look", + "err_mods_empty", + "err_payment_currency", + "err_payment_exchangerates", + "err_signature", + "err_api_action", + "err_api_callback", + "err_api_ip", + "user_not_found", + "wait_info", + "order_id_duplicate", + "shop_blocked", + "shop_not_active", + "5", + ], + LiqpayExpireErrcode, + LiqpayRequestErrcode, + LiqpayPaymentErrcode, +] + +LiqpayFinancialErrcode = _Literal[ + "90", + "101", + "102", + "103", + "104", + "105", + "106", + "107", + "108", + "109", + "110", + "111", + "112", + "113", + "114", + "115", + "2903", + "2915", + "3914", + "9851", + "9852", + "9854", + "9855", + "9857", + "9859", + "9860", + "9861", + "9863", + "9867", + "9868", + "9872", + "9882", + "9886", + "9961", + "9989", +] + +LiqPayErrcode = _Union[ + UnknownErrcode, + LiqpayAntiFraudErrcode, + LiqpayFinancialErrcode, + LiqpayNonFinancialErrcode, +] diff --git a/liqpy/types/post.py b/liqpy/types/post.py new file mode 100644 index 0000000..83206f5 --- /dev/null +++ b/liqpy/types/post.py @@ -0,0 +1,20 @@ +from typing import Mapping, Union, Literal, Callable, List, TypedDict +from requests import Response + + +Proxies = Mapping[str, str] +Timeout = Union[float, tuple[float, float]] +Hook = Callable[[Response], None] +Hooks = Mapping[Literal["response"], Union[Hook, List[Hook]]] +Verify = Union[bool, str] +Cert = Union[str, tuple[str, str]] + + +class PostParams(TypedDict, total=False): + stream: bool + allow_redirects: bool + proxies: Proxies + timeout: Timeout + hooks: Hooks + verify: Verify + cert: Cert diff --git a/liqpy/types/request.py b/liqpy/types/request.py new file mode 100644 index 0000000..c960e29 --- /dev/null +++ b/liqpy/types/request.py @@ -0,0 +1,122 @@ +from typing import Literal, TypedDict, Required, List +from numbers import Number +from datetime import datetime +from uuid import UUID + +from liqpy.data import FiscalItem, DetailAddenda, SplitRule, FiscalInfo + +from .common import SubscribePeriodicity, Language, PayType, PayOption, Format +from .action import Action + + +class DetailAddendaDict(TypedDict): + airLine: str + ticketNumber: str + passengerName: str + flightNumber: str + originCity: str + destinationCity: str + departureDate: str + + +class SplitRuleDict(TypedDict, total=False): + public_key: Required[str] + amount: Required[Number] + commission_payer: Required[Literal["sender", "receiver"]] + server_url: Required[str] + + +class FiscalItemDict(TypedDict): + id: int + amount: Number + cost: Number + price: Number + + +class FiscalInfoDict(TypedDict, total=False): + items: List[FiscalItemDict | FiscalItem] + delivery_emails: List[str] + + +class ProductDict(TypedDict, total=False): + product_category: str + product_description: str + product_name: str + product_url: str + + +class OneClickDict(TypedDict, total=False): + customer: str + recurringbytoken: Literal["1", True] + customer_user_id: str + + +class SubscribeDict(TypedDict, total=False): + subscribe: Literal[1, True] + subscribe_date_start: str | datetime + subscribe_periodicity: SubscribePeriodicity + + +class LetterDict(TypedDict, total=False): + letter_of_credit: Literal[1, True] + letter_of_credit_date: str | datetime + + +class MPIParamsDict(TypedDict, total=False): + mpi_md: str + mpi_pares: str + + +class SenderDict(TypedDict, total=False): + phone: str + sender_phone: str + sender_first_name: str + sender_last_name: str + sender_email: str + sender_country_code: str + sender_city: str + sender_address: str + sender_postal_code: str + sender_shipping_state: str + + +class CardDict(TypedDict, total=False): + card: Required[str] + card_exp_month: str + card_exp_year: str + card_cvv: str + + +class BaseRequestDict(TypedDict, total=False): + version: Required[int] + public_key: Required[str] + action: Required["Action"] + + +class LiqpayRequestDict( + CardDict, + SenderDict, + MPIParamsDict, + LetterDict, + SubscribeDict, + OneClickDict, + ProductDict, + total=False, +): + order_id: str | UUID + amount: Number | str + rro_info: FiscalInfoDict | FiscalInfo + expired_date: str | datetime + language: Language + paytype: PayType + paytypes: PayOption + result_url: str + server_url: str + verifycode: Literal["Y", True] + email: str + date_from: int | datetime + date_to: int | datetime + resp_format: Format + split_rules: list[SplitRuleDict | SplitRule] + dae: DetailAddendaDict | DetailAddenda + info: str diff --git a/liqpy/types/status.py b/liqpy/types/status.py new file mode 100644 index 0000000..a1fbe4d --- /dev/null +++ b/liqpy/types/status.py @@ -0,0 +1,40 @@ +from typing import Literal as _Literal, Union as _Union + + +SubscriptionStatus = _Literal["subscribed", "unsubscribed"] +ErrorStatus = _Literal["error", "failure"] +SuccessStatus = _Literal["success"] +FinalStatus = _Union[ + ErrorStatus, _Literal["reversed"], SubscriptionStatus, SuccessStatus +] +ConfirmationStatus = _Literal[ + "3ds_verify", + "captcha_verify", + "cvv_verify", + "ivr_verify", + "otp_verify", + "password_verify", + "phone_verify", + "pin_verify", + "receiver_verify", + "sender_verify", + "senderapp_verify", + "wait_qr", + "wait_sender", +] +OtherStatus = _Literal[ + "cash_wait", + "hold_wait", + "invoice_wait", + "prepared", + "processing", + "wait_accept", + "wait_card", + "wait_compensation", + "wait_lc", + "wait_reserve", + "wait_secure", +] +Status = CallbackStatus = _Union[ + SubscriptionStatus, FinalStatus, ConfirmationStatus, OtherStatus +] diff --git a/liqpy/util.py b/liqpy/util.py deleted file mode 100644 index 0c6680f..0000000 --- a/liqpy/util.py +++ /dev/null @@ -1,95 +0,0 @@ -from typing import overload, TYPE_CHECKING -from functools import singledispatch -from numbers import Number -from datetime import datetime -from urllib.parse import urlparse - - -if TYPE_CHECKING: - from .types import RequestForm - - -def to_dict(data: str, signature: str, /) -> "RequestForm": - """Convert data and signature into a dictionary.""" - return {"data": data, "signature": signature} - - -def filter_none(data: dict, /) -> dict: - return {key: value for key, value in data.items() if value is not None} - - -def is_sandbox(key: str, /) -> bool: - return key.startswith("sandbox_") - - -def verify_url(value: str): - assert len(value) <= 500, "Invalid URL. Must be less than 500 characters." - result = urlparse(value or "") - assert result.scheme in ("http", "https"), "Invalid URL scheme. Must be http or https." - assert result.netloc != "", "Invalid URL. Must be a valid URL." - - -@singledispatch -def to_milliseconds(value) -> int: - raise NotImplementedError(f"Unsupported type: {type(value)}") - -@to_milliseconds.register -def _(value: Number): - return int(round(value * 1000)) - - -@to_milliseconds.register -def _(value: datetime): - return to_milliseconds(value.timestamp()) - - -@to_milliseconds.register -def _(value: str): - return to_milliseconds(datetime.fromisoformat(value)) - - -@singledispatch -def format_date(value) -> str: - raise NotImplementedError(f"Unsupported type: {type(value)}") - - -@format_date.register -def _(value: datetime): - return value.strftime("%Y-%m-%d %H:%M:%S") - - -@format_date.register -def _(value: str): - return format_date(datetime.fromisoformat(value)) - - -@format_date.register -def _(value: Number): - return format_date(datetime.fromtimestamp(value)) - - -if TYPE_CHECKING: - - @overload - def to_milliseconds(value: Number) -> int: - ... - - @overload - def to_milliseconds(value: datetime) -> int: - ... - - @overload - def to_milliseconds(value: str) -> int: - ... - - @overload - def format_date(value: Number) -> str: - ... - - @overload - def format_date(value: datetime) -> str: - ... - - @overload - def format_date(value: str) -> str: - ... diff --git a/liqpy/validation.py b/liqpy/validation.py new file mode 100644 index 0000000..593d329 --- /dev/null +++ b/liqpy/validation.py @@ -0,0 +1,311 @@ +from typing import TYPE_CHECKING +from functools import cache +from datetime import datetime +from re import compile +from numbers import Number +from uuid import UUID +from urllib.parse import urlparse + +from .data import DetailAddenda, SplitRule, FiscalItem, FiscalInfo + +if TYPE_CHECKING: + from .types.request import ( + DetailAddendaDict, + SplitRuleDict, + FiscalItemDict, + FiscalInfoDict, + LiqpayRequestDict, + ) + + +@cache +def phone_pattern(): + return compile(r"\+?380\d{9}") + + +@cache +def card_cvv_pattern(): + return compile(r"\d{3}") + + +@cache +def card_number_pattern(): + return compile(r"\d{16}") + + +@cache +def card_exp_year_pattern(): + return compile(r"(\d{2})?\d{2}") + + +@cache +def card_exp_month_pattern(): + return compile(r"(0[1-9])|(1[0-2])") + + +def noop(value, /, **kwargs): + pass + + +def number(value, /): + assert isinstance(value, Number), f"value must be a number" + + +def gt(value, /, *, threshold: Number): + number(value) + assert value > threshold, f"value must be greater than {threshold}" + + +def string(value, /, *, max_len: int | None = None): + assert isinstance(value, str), f"value must be a string" + if max_len is not None: + assert len(value) <= max_len, f"string must be less than {max_len} characters" + + +def url(value, /, *, max_len: int | None = None): + string(value, max_len=max_len) + result = urlparse(value or "") + assert result.scheme in ( + "http", + "https", + ), "Invalid URL scheme. Must be http or https." + assert result.netloc != "", f"Must be a valid URL." + + +def to_dict(o: dict[str], cls: type) -> dict: + if isinstance(o, cls): + return o.__dict__ + elif isinstance(o, dict): + return o + else: + raise AssertionError(f"Invalid object type. Must be {cls} or dict.") + + +def check_required(params: dict[str], keys: set[str]): + missing = keys - params.keys() + assert not missing, f"Missing required parameters: {missing}" + + +class BaseValidator: + def __call__(self, o: "LiqpayRequestDict", /, **kwargs): + for key, value in o.items(): + try: + getattr(self, key, noop)(value, **kwargs.get(key, {})) + except AssertionError as e: + raise AssertionError(f"Invalid {key} parameter.") from e + except Exception as e: + raise Exception(f"Failed to verify {key} parameter.") from e + + +class Validator(BaseValidator): + def version(self, value, /, **kwargs): + assert isinstance(value, int), f"version must be an integer" + assert value > 0, f"version must be greater than 0" + + def public_key(self, value, /, **kwargs): + string(value) + + def action(self, value, /, **kwargs): + string(value) + + def order_id(self, value, /, **kwargs): + if not isinstance(value, UUID): + assert isinstance(value, str), "order_id must be a string or UUID" + assert len(value) <= 255, "order_id must be less than 255 characters" + + def description(self, value, /, **kwargs): + # NOTE: API allows to request up to 49 720 characters, + # but cuts to 2048 characters long + string(value, max_len=2048) + + def amount(self, value, /, **kwargs): + gt(value, threshold=0) + + def currency(self, value, /, **kwargs): + assert value in ("USD", "UAH"), "currency must be USD or UAH" + + def expired_date(self, value, /, **kwargs): + assert isinstance(value, datetime), "expired_date must be a datetime" + + def date_from(self, value, /, **kwargs): + assert isinstance(value, int), "date_from must be a int" + + def date_to(self, value, /, **kwargs): + assert isinstance(value, int), "date_to must be a int" + + def resp_format(self, value, /, **kwargs): + assert value in ( + "json", + "csv", + "xml", + ), "format must be json, csv or xml" + + def phone(self, value, /, **kwargs): + assert phone_pattern().fullmatch( + value + ), "phone must be in format +380XXXXXXXXX or 380XXXXXXXXX" + + def sender_phone(self, value, /, **kwargs): + self.phone(value) + + def info(self, value, /, **kwargs): + string(value) + + def language(self, value, /, **kwargs): + assert value in ("uk", "en"), "language must be uk or en" + + def card_number(self, value, /, **kwargs): + assert card_number_pattern().fullmatch(value), f"card must be 16 digits long" + + def card_cvv(self, value, /, **kwargs): + assert card_cvv_pattern().fullmatch(value), f"cvv must be 3 digits long" + + def card_exp_year(self, value, /, **kwargs): + assert card_exp_year_pattern().fullmatch( + value + ), f"exp_year must be 2 or 4 digits long" + + def card_exp_month(self, value, /, **kwargs): + assert card_exp_month_pattern().fullmatch( + value + ), f"exp_month must be 2 digits long and between 01 and 12" + + def subscribe(self, value, /, **kwargs): + assert value == 1, "subscribe must be 1" + + def subscribe_periodicity(self, value, /, **kwargs): + assert value in ("month", "year"), "subscribe_periodicity must be month or year" + + def subscribe_date_start(self, value, /, **kwargs): + assert isinstance(value, datetime), "subscribe_date_start must be a datetime" + + def paytype(self, value, /, **kwargs): + assert value in ( + "apay", + "gpay", + "apay_tavv", + "gpay_tavv", + "tavv", + ), "paytype must be one of: apay, gpay, apay_tavv, gpay_tavv, tavv" + + def payoption(self, value, /, **kwargs): + assert value in ( + "apay", + "gpay", + "card", + "liqpay", + "moment_part", + "paypart", + "cash", + "invoice", + "qr", + ), "paytypes must be one of: apay, gpay, card, liqpay, moment_part, paypart, cash, invoice, qr" + + def paytypes(self, value, /, **kwargs): + if isinstance(value, list): + for i, item in enumerate(value): + try: + self.payoption(item, **kwargs) + except AssertionError as e: + raise AssertionError(f"Invalid paytypes element {i}.") from e + + def customer(self, value, /, **kwargs): + string(value, max_len=100) + + def customer_user_id(self, value, /, **kwargs): + string(value) + + def reccuringbytoken(self, value, /, **kwargs): + assert value == "1", "reccuringbytoken must be 1" + + def dae(self, value, /, **kwargs): + value: "DetailAddendaDict" = to_dict(value, DetailAddenda) + + try: + string(value["air_line"], max_len=4) + string(value["ticket_number"], max_len=15) + string(value["passenger_name"], max_len=29) + string(value["flight_number"], max_len=5) + string(value["origin_city"], max_len=5) + string(value["destination_city"], max_len=5) + except AssertionError as e: + raise AssertionError("Invalid dae object.") from e + + def split_rule(self, value, /, **kwargs): + value: "SplitRuleDict" = to_dict(value, SplitRule) + + string(value["public_key"]) + gt(value["amount"], threshold=0) + assert value["commission_payer"] in ( + "sender", + "receiver", + ), "commission_payer must be sender or receiver" + url(value["server_url"]) + + def split_rules(self, value, /, **kwargs): + assert isinstance(value, list), "split_rules must be a list" + for i, rule in enumerate(value): + try: + self.split_rule(rule, **kwargs) + except AssertionError as e: + raise AssertionError( + f"Invalid split_rule[{i}] object in split_rules." + ) from e + + def fiscal_data(self, value, /, **kwargs): + value: "FiscalItemDict" = to_dict(value, FiscalItem) + + assert isinstance(value["id"], int), "Invalid id. Must be an integer." + + for value in (value["amount"], value["cost"], value["price"]): + gt(value, threshold=0) + + def rro_info(self, value, /, **kwargs): + value: "FiscalInfoDict" = to_dict(value, FiscalInfo) + + assert isinstance(value["items"], list), "items in rro_info must be a list" + assert isinstance( + value["delivery_emails"], list + ), "delivery_emails in rro_info must be a list" + + for i, item in enumerate(value["items"]): + try: + self.fiscal_data(item, **kwargs) + except AssertionError as e: + raise AssertionError(f"Invalid items[{i}] object in rro_info.") from e + + for i, email in enumerate(value.delivery_emails): + try: + string(email) + except AssertionError as e: + raise AssertionError( + f"Invalid delivery_emails[{i}] object in rro_info." + ) from e + + def verifycode(self, value, /, **kwargs): + assert value == "Y", "verifycode must be Y" + + def server_url(self, value, /, **kwargs): + url(value, max_len=510) + + def result_url(self, value, /, **kwargs): + # string(value, max_len=510) + url(value, max_len=510) + + def product_url(self, value, /, **kwargs): + url(value, max_len=2000) + + def product_description(self, value, /, **kwargs): + string(value, max_len=500) + + def product_name(self, value, /, **kwargs): + string(value, max_len=100) + + def product_category(self, value, /, **kwargs): + string(value, max_len=25) + + def product_name(self, value, /, **kwargs): + string(value, max_len=100) + + def product_category(self, value, /, **kwargs): + string(value, max_len=25) diff --git a/pyproject.toml b/pyproject.toml index 02944dd..242c69d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,5 +19,5 @@ Repository = "https://github.com/rostyq/liqpy" [tool.hatch.version] path = "liqpy/__init__.py" -[tool.hatch.metadata.hooks.requirements-txt] +[tool.hatch.metadata.hooks.requirements_txt] files = ["requirements.txt"] diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..e69de29 diff --git a/readme.ipynb b/readme.ipynb index a0757a6..648ec92 100644 --- a/readme.ipynb +++ b/readme.ipynb @@ -7,6 +7,13 @@ "# LiqPay API Notebook" ] }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Setup" + ] + }, { "cell_type": "markdown", "metadata": {}, @@ -24,7 +31,7 @@ "%autoreload 2\n", "\n", "%load_ext dotenv\n", - "%dotenv" + "%dotenv -o" ] }, { @@ -41,15 +48,12 @@ "outputs": [], "source": [ "import json\n", - "import random\n", - "import string\n", "\n", + "from os import getenv\n", "from uuid import uuid4\n", "from pprint import pprint\n", "from datetime import date, datetime, timedelta, UTC\n", "\n", - "from liqpy.client import Client\n", - "from liqpy.exceptions import LiqPayException\n", "from liqpy.testing import TestCard, gen_card_cvv, gen_card_expire" ] }, @@ -57,16 +61,7 @@ "cell_type": "markdown", "metadata": {}, "source": [ - "Get URL for callbacks." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "%env SERVER_URL" + "Initialize client with public and private keys" ] }, { @@ -75,24 +70,17 @@ "metadata": {}, "outputs": [], "source": [ - "SERVER_URL = _" + "from liqpy.client import Client\n", + "\n", + "client = Client()\n", + "client" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "Initialize client with public and private keys" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "client = Client()\n", - "client" + "## Examples" ] }, { @@ -108,20 +96,19 @@ "metadata": {}, "outputs": [], "source": [ - "data, signature = client.encode(\n", + "params, signature = client.encode(\n", " action=\"pay\",\n", " amount=1,\n", - " order_id=str(uuid4()),\n", + " order_id=uuid4(),\n", " description=\"Test Encoding\",\n", " currency=\"USD\",\n", - " server_url=SERVER_URL,\n", ")\n", "\n", "sep, end = \"\\n\", \"\\n\\n\"\n", - "print(\"data:\", data, sep=sep, end=end)\n", + "print(\"data:\", params, sep=sep, end=end)\n", "print(\"signature:\", signature, sep=sep, end=end)\n", "\n", - "client.callback(data, signature)" + "client.callback(params, signature)" ] }, { @@ -139,9 +126,8 @@ "source": [ "card_exp_month, card_exp_year = gen_card_expire(valid=True)\n", "\n", - "order_id = str(uuid4())\n", - "client.request(\n", - " \"pay\",\n", + "order_id = uuid4()\n", + "client.pay(\n", " order_id=order_id,\n", " amount=1,\n", " currency=\"USD\",\n", @@ -150,10 +136,25 @@ " card_exp_month=card_exp_month,\n", " card_exp_year=card_exp_year,\n", " card_cvv=gen_card_cvv(),\n", - " server_url=SERVER_URL,\n", ")" ] }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Refund the payment" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "client.refund(order_id=order_id, amount=1)" + ] + }, { "cell_type": "markdown", "metadata": {}, @@ -167,12 +168,13 @@ "metadata": {}, "outputs": [], "source": [ + "from liqpy.exceptions import LiqPayException\n", + "\n", "try:\n", " card_exp_month, card_exp_year = gen_card_expire(valid=True)\n", "\n", " order_id = str(uuid4())\n", - " client.request(\n", - " \"pay\",\n", + " client.pay(\n", " order_id=order_id,\n", " amount=1,\n", " currency=\"USD\",\n", @@ -181,9 +183,9 @@ " card_exp_month=card_exp_month,\n", " card_exp_year=card_exp_year,\n", " card_cvv=gen_card_cvv(),\n", - " server_url=SERVER_URL,\n", " )\n", "except LiqPayException as e:\n", + " print(e.code, e)\n", " print(e.response)\n", " pprint(e.details)\n", " raise e" @@ -218,14 +220,14 @@ "metadata": {}, "outputs": [], "source": [ - "client.data(order_id, \"Test info\")" + "client.data(order_id, \"Lorem Ipsum\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "Create checkout link" + "Test checkout callback" ] }, { @@ -234,14 +236,37 @@ "metadata": {}, "outputs": [], "source": [ - "client.checkout(\n", - " action=\"pay\",\n", + "from liqpy.server import LiqpayServer\n", + "from webbrowser import open_new_tab\n", + "\n", + "action = \"auth\"\n", + "order_id = uuid4()\n", + "\n", + "expire = timedelta(seconds=20)\n", + "timeout = 10 \n", + "timeout = (expire + timedelta(seconds=timeout)).total_seconds()\n", + "\n", + "server_url = getenv(\"SERVER_URL\") or None\n", + "\n", + "checkout_url = client.checkout(\n", + " action,\n", " amount=1,\n", - " order_id=str(uuid4()),\n", - " description=\"Test Checkout\",\n", + " order_id=uuid4(),\n", + " description=f\"test {action} checkout\",\n", " currency=\"USD\",\n", - " server_url=SERVER_URL\n", - ")" + " expired_date=expire,\n", + " # subscribe_date_start=timedelta(days=7),\n", + " # subscribe_periodicity=\"month\",\n", + " result_url=\"https://example.com/result\",\n", + " server_url=server_url,\n", + ")\n", + "\n", + "print(\"checkout link\\n\", checkout_url)\n", + "open_new_tab(checkout_url)\n", + "\n", + "if server_url is not None:\n", + " with LiqpayServer(client=client, timeout=timeout) as server:\n", + " server.handle_request()" ] }, { @@ -258,10 +283,69 @@ "outputs": [], "source": [ "date_to = datetime.now(UTC)\n", - "date_from = date_to - timedelta(days=30)\n", + "date_from = date_to - timedelta(days=1)\n", + "print(\"from:\", date_from)\n", + "print(\"to:\", date_to)\n", + "\n", + "params = client.reports(date_from=date_from, date_to=date_to, format=\"json\")\n", + "params = json.loads(params)\n", + "params" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "print(client.reports(date_from=date_from, date_to=date_to, format=\"csv\"))" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "print(client.reports(date_from=date_from, date_to=date_to, format=\"xml\"))" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Create subscription" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "card_exp_month, card_exp_year = gen_card_expire(valid=True)\n", "\n", - "data = client.reports(date_from=date_from, date_to=date_to, format=\"json\")\n", - "data = json.loads(data)" + "order_id = uuid4()\n", + "client.subscribe(\n", + " amount=1,\n", + " order_id=order_id,\n", + " description=\"Test Subscribe\",\n", + " currency=\"USD\",\n", + " card=TestCard.successful(),\n", + " card_exp_month=card_exp_month,\n", + " card_exp_year=card_exp_year,\n", + " card_cvv=gen_card_cvv(),\n", + " # phone=\"+380661234567\",\n", + " subscribe_periodicity=\"month\",\n", + " subscribe_date_start=timedelta()\n", + ")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Unsubscribe" ] }, { @@ -269,7 +353,9 @@ "execution_count": null, "metadata": {}, "outputs": [], - "source": [] + "source": [ + "client.unsubscribe(order_id=order_id)" + ] } ], "metadata": { diff --git a/tests/test_json_encoder.py b/tests/test_json_encoder.py new file mode 100644 index 0000000..972c5b7 --- /dev/null +++ b/tests/test_json_encoder.py @@ -0,0 +1,67 @@ +from datetime import datetime, date, UTC, timezone, timedelta +from uuid import UUID +from decimal import Decimal + +from pytest import fixture + +from liqpy.api import LiqPayJSONEncoder, encode +from liqpy.util import DetailAddenda + + +@fixture +def encoder(): + return LiqPayJSONEncoder() + + +def test_encode_bytes(encoder: LiqPayJSONEncoder): + assert encoder.encode(b"") == '""' + assert encoder.encode(b"test") == '"test"' + + +def test_encode_date(encoder: LiqPayJSONEncoder): + assert encoder.encode(date(2021, 1, 2)) == '"2021-01-02 00:00:00"' + + +def test_encode_datetime(encoder: LiqPayJSONEncoder): + assert ( + encoder.encode(datetime(2021, 1, 2, 3, 4, 5, tzinfo=UTC)) + == '"2021-01-02 03:04:05"' + ) + assert ( + encoder.encode( + datetime(2021, 1, 2, 3, 4, 5, tzinfo=timezone(timedelta(hours=3))) + ) + == '"2021-01-02 00:04:05"' + ) + assert ( + encoder.encode( + datetime(2021, 1, 2, 3, 4, 5, tzinfo=timezone(timedelta(hours=-3))) + ) + == '"2021-01-02 06:04:05"' + ) + + +def test_encode_uuid(encoder: LiqPayJSONEncoder): + value = "123e4567-e89b-12d3-a456-426614174000" + assert encoder.encode(UUID(value)) == f'"{value}"' + + +def test_encode_decimal(encoder: LiqPayJSONEncoder): + assert encoder.encode(Decimal(1)) == "1.0" + assert encoder.encode(Decimal("1.0")) == "1.0" + assert encoder.encode(Decimal(1.0)) == "1.0" + assert encoder.encode(Decimal("0.0001")) == "0.0001" + + +def test_encode_dae(encoder: LiqPayJSONEncoder): + dae = DetailAddenda( + air_line="Avia", + ticket_number="ACSFD12354SA", + passenger_name="John Doe", + flight_number="742", + origin_city="DP", + destination_city="NY", + departure_date=date(2014, 5, 10), + ) + # TODO: verify + encoder.encode(dae)