Skip to content

Payment API (experimental alpha)

jukto.interfaces.payments.BasePaymentProvider

Bases: TransportLifecycle, ABC

Source code in src/jukto/interfaces/payments.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class BasePaymentProvider(TransportLifecycle, ABC):
    capabilities: PaymentCapabilities

    @abstractmethod
    def initiate(self, request: PaymentRequest) -> PaymentInitiationResult: ...
    @abstractmethod
    def verify(
        self, validation_id: str, expected: PaymentReference
    ) -> PaymentVerificationResult: ...
    @abstractmethod
    def query_transaction(self, expected: PaymentReference) -> PaymentQueryResult: ...
    @abstractmethod
    def query_session(
        self, session_id: str, expected: PaymentReference
    ) -> PaymentVerificationResult: ...
    @abstractmethod
    def verify_notification(
        self, notification: Mapping[str, str], expected: PaymentReference
    ) -> PaymentVerificationResult | PaymentQueryResult: ...
    @abstractmethod
    def refund(self, request: RefundRequest) -> RefundResult: ...
    @abstractmethod
    def query_refund(
        self, provider_refund_id: str, bank_transaction_id: str
    ) -> RefundResult: ...

jukto.interfaces.payments.AsyncBasePaymentProvider

Bases: AsyncTransportLifecycle, ABC

Source code in src/jukto/interfaces/payments.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class AsyncBasePaymentProvider(AsyncTransportLifecycle, ABC):
    capabilities: PaymentCapabilities

    @abstractmethod
    async def initiate(self, request: PaymentRequest) -> PaymentInitiationResult: ...
    @abstractmethod
    async def verify(
        self, validation_id: str, expected: PaymentReference
    ) -> PaymentVerificationResult: ...
    @abstractmethod
    async def query_transaction(
        self, expected: PaymentReference
    ) -> PaymentQueryResult: ...
    @abstractmethod
    async def query_session(
        self, session_id: str, expected: PaymentReference
    ) -> PaymentVerificationResult: ...
    @abstractmethod
    async def verify_notification(
        self, notification: Mapping[str, str], expected: PaymentReference
    ) -> PaymentVerificationResult | PaymentQueryResult: ...
    @abstractmethod
    async def refund(self, request: RefundRequest) -> RefundResult: ...
    @abstractmethod
    async def query_refund(
        self, provider_refund_id: str, bank_transaction_id: str
    ) -> RefundResult: ...

jukto.providers.payments.sslcommerz.SSLCommerzClient

Bases: SSLCommerzContract, BasePaymentProvider

Source code in src/jukto/providers/payments/sslcommerz.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
class SSLCommerzClient(SSLCommerzContract, BasePaymentProvider):
    def __init__(
        self,
        store_id: str,
        store_password: str,
        *,
        sandbox: bool = True,
        timeout: float | httpx.Timeout = 30.0,
        http_client: httpx.Client | None = None,
        limits: httpx.Limits | None = None,
    ) -> None:
        for value in (store_id, store_password):
            if not isinstance(value, str) or not value.strip() or len(value) > 30:
                raise JuktoConfigurationError(
                    "SSLCommerz requires nonempty store credentials of at most 30 characters"
                )
        if type(sandbox) is not bool:
            raise JuktoConfigurationError("sandbox must be a boolean")
        self.store_id, self.store_password = store_id, store_password
        self.base_url = (
            "https://sandbox.sslcommerz.com"
            if sandbox
            else "https://securepay.sslcommerz.com"
        )
        validate_endpoint(self.base_url)
        self._transport = SyncTransport(
            timeout=timeout, http_client=http_client, limits=limits
        )

    def _request(
        self, method: str, path: str, *, side_effect: bool = False, **kwargs: Any
    ) -> httpx.Response:
        self._transport.ensure_open()
        if side_effect:
            # Provider refund initiation is GET but still changes financial state.
            before_request("POST", self.base_url + path)
        try:
            with safe_httpx_logs():
                return self._transport.request(
                    method,
                    self.base_url + path,
                    headers={"Cache-Control": "no-store"},
                    **kwargs,
                )
        except httpx.TimeoutException:
            raise ProviderTimeoutError(
                "SSLCommerz request timed out; details withheld",
                provider=self.provider_name,
            ) from None
        except httpx.RequestError:
            raise ProviderConnectionError(
                "SSLCommerz connection failed; details withheld",
                provider=self.provider_name,
            ) from None

    @operation
    def initiate(self, request: PaymentRequest) -> PaymentInitiationResult:
        payload = self._prepare_initiation(request)
        return self._initiation(
            self._request("POST", self.INITIATE, data=payload), request.reference
        )

    @operation
    def verify(
        self, validation_id: str, expected: PaymentReference
    ) -> PaymentVerificationResult:
        text(validation_id, "validation_id", 50)
        response = self._request(
            "GET", self.VALIDATE, params=dict(self._credentials(), val_id=validation_id)
        )
        return self._validated(response, expected, validation_id)

    @operation
    def query_transaction(self, expected: PaymentReference) -> PaymentQueryResult:
        response = self._request(
            "GET",
            self.TRANSACTION,
            params=dict(self._credentials(), tran_id=expected.merchant_transaction_id),
        )
        return self._query(response, expected)

    @operation
    def query_session(
        self, session_id: str, expected: PaymentReference
    ) -> PaymentVerificationResult:
        text(session_id, "session_id", 50)
        response = self._request(
            "GET",
            self.TRANSACTION,
            params=dict(self._credentials(), sessionkey=session_id),
        )
        return self._session(response, expected, session_id)

    def verify_notification(
        self, notification: Mapping[str, str], expected: PaymentReference
    ) -> PaymentVerificationResult | PaymentQueryResult:
        validation_id = self._notification(notification, expected)
        return (
            self.verify(validation_id, expected)
            if validation_id is not None
            else self.query_transaction(expected)
        )

    @operation
    def refund(self, request: RefundRequest) -> RefundResult:
        payload = self._refund_payload(request)
        response = self._request(
            "GET", self.TRANSACTION, params=payload, side_effect=True
        )
        return self._refund(response, request.bank_transaction_id, None, request)

    @operation
    def query_refund(
        self, provider_refund_id: str, bank_transaction_id: str
    ) -> RefundResult:
        text(provider_refund_id, "provider_refund_id", 50)
        text(bank_transaction_id, "bank_transaction_id", 80)
        response = self._request(
            "GET",
            self.TRANSACTION,
            params=dict(self._credentials(), refund_ref_id=provider_refund_id),
        )
        return self._refund(response, bank_transaction_id, provider_refund_id)

jukto.providers.payments.async_sslcommerz.AsyncSSLCommerzClient

Bases: SSLCommerzContract, AsyncBasePaymentProvider

Source code in src/jukto/providers/payments/async_sslcommerz.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
class AsyncSSLCommerzClient(SSLCommerzContract, AsyncBasePaymentProvider):
    def __init__(
        self,
        store_id: str,
        store_password: str,
        *,
        sandbox: bool = True,
        timeout: float | httpx.Timeout = 30.0,
        http_client: httpx.AsyncClient | None = None,
        limits: httpx.Limits | None = None,
    ) -> None:
        for value in (store_id, store_password):
            if not isinstance(value, str) or not value.strip() or len(value) > 30:
                raise JuktoConfigurationError(
                    "SSLCommerz requires nonempty store credentials of at most 30 characters"
                )
        if type(sandbox) is not bool:
            raise JuktoConfigurationError("sandbox must be a boolean")
        self.store_id, self.store_password = store_id, store_password
        self.base_url = (
            "https://sandbox.sslcommerz.com"
            if sandbox
            else "https://securepay.sslcommerz.com"
        )
        validate_endpoint(self.base_url)
        self._transport = AsyncTransport(
            timeout=timeout, http_client=http_client, limits=limits
        )

    async def _request(
        self, method: str, path: str, *, side_effect: bool = False, **kwargs: Any
    ) -> httpx.Response:
        self._transport.ensure_open()
        if side_effect:
            # Provider refund initiation is GET but still changes financial state.
            before_request("POST", self.base_url + path)
        try:
            with safe_httpx_logs():
                return await self._transport.request(
                    method,
                    self.base_url + path,
                    headers={"Cache-Control": "no-store"},
                    **kwargs,
                )
        except httpx.TimeoutException:
            raise ProviderTimeoutError(
                "SSLCommerz request timed out; details withheld",
                provider=self.provider_name,
            ) from None
        except httpx.RequestError:
            raise ProviderConnectionError(
                "SSLCommerz connection failed; details withheld",
                provider=self.provider_name,
            ) from None

    @operation
    async def initiate(self, request: PaymentRequest) -> PaymentInitiationResult:
        payload = self._prepare_initiation(request)
        return self._initiation(
            await self._request("POST", self.INITIATE, data=payload), request.reference
        )

    @operation
    async def verify(
        self, validation_id: str, expected: PaymentReference
    ) -> PaymentVerificationResult:
        text(validation_id, "validation_id", 50)
        response = await self._request(
            "GET", self.VALIDATE, params=dict(self._credentials(), val_id=validation_id)
        )
        return self._validated(response, expected, validation_id)

    @operation
    async def query_transaction(self, expected: PaymentReference) -> PaymentQueryResult:
        response = await self._request(
            "GET",
            self.TRANSACTION,
            params=dict(self._credentials(), tran_id=expected.merchant_transaction_id),
        )
        return self._query(response, expected)

    @operation
    async def query_session(
        self, session_id: str, expected: PaymentReference
    ) -> PaymentVerificationResult:
        text(session_id, "session_id", 50)
        response = await self._request(
            "GET",
            self.TRANSACTION,
            params=dict(self._credentials(), sessionkey=session_id),
        )
        return self._session(response, expected, session_id)

    async def verify_notification(
        self, notification: Mapping[str, str], expected: PaymentReference
    ) -> PaymentVerificationResult | PaymentQueryResult:
        validation_id = self._notification(notification, expected)
        return (
            await self.verify(validation_id, expected)
            if validation_id is not None
            else await self.query_transaction(expected)
        )

    @operation
    async def refund(self, request: RefundRequest) -> RefundResult:
        payload = self._refund_payload(request)
        response = await self._request(
            "GET", self.TRANSACTION, params=payload, side_effect=True
        )
        return self._refund(response, request.bank_transaction_id, None, request)

    @operation
    async def query_refund(
        self, provider_refund_id: str, bank_transaction_id: str
    ) -> RefundResult:
        text(provider_refund_id, "provider_refund_id", 50)
        text(bank_transaction_id, "bank_transaction_id", 80)
        response = await self._request(
            "GET",
            self.TRANSACTION,
            params=dict(self._credentials(), refund_ref_id=provider_refund_id),
        )
        return self._refund(response, bank_transaction_id, provider_refund_id)

jukto.payments.PaymentCapabilities dataclass

Source code in src/jukto/payments.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
@dataclass(frozen=True)
class PaymentCapabilities:
    hosted_checkout: bool = True
    server_verification: bool = True
    notification_verification: bool = True
    transaction_query: bool = True
    partial_refund: bool = True
    refund_query: bool = True
    separate_execution: bool = False
    provider_enforced_idempotency: bool | None = (
        None  # Unverified; references are not guarantees.
    )
    supported_currencies: tuple[str, ...] = ("BDT",)

jukto.payments.PaymentReference dataclass

Source code in src/jukto/payments.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@dataclass(frozen=True)
class PaymentReference:
    merchant_transaction_id: str
    amount: Decimal
    currency: str

    def __post_init__(self) -> None:
        text(self.merchant_transaction_id, "merchant_transaction_id", 30)
        amount(self.amount)
        if self.currency != "BDT":
            raise JuktoValidationError(
                "This integration currently supports explicit BDT only",
                provider="SSLCommerz",
            )

jukto.payments.PaymentRequest dataclass

Source code in src/jukto/payments.py 
106
107
108
109
110
111
112
@dataclass(frozen=True)
class PaymentRequest:
    reference: PaymentReference
    customer: PaymentCustomer
    urls: PaymentURLs
    product_name: str
    product_category: str

jukto.payments.PaymentCustomer dataclass

Source code in src/jukto/payments.py 
82
83
84
85
86
87
88
89
90
@dataclass(frozen=True)
class PaymentCustomer:
    name: str
    email: str
    phone: str
    address: str
    city: str
    postcode: str
    country: str

jukto.payments.PaymentURLs dataclass

Source code in src/jukto/payments.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@dataclass(frozen=True)
class PaymentURLs:
    success: str
    failure: str
    cancel: str
    ipn: str

    def __post_init__(self) -> None:
        for value in (self.success, self.failure, self.cancel, self.ipn):
            text(value, "callback URL", 255)
            validate_endpoint(value)

jukto.payments.PaymentStatus

Bases: str, Enum

Source code in src/jukto/payments.py 
12
13
14
15
16
17
18
class PaymentStatus(str, Enum):
    REQUIRES_ACTION = "requires_action"
    PENDING = "pending"
    PAID = "paid"
    FAILED = "failed"
    CANCELLED = "cancelled"
    UNKNOWN = "unknown"

jukto.payments.PaymentInitiationResult dataclass

Source code in src/jukto/payments.py 
115
116
117
118
119
120
121
122
123
124
125
126
@dataclass(frozen=True)
class PaymentInitiationResult:
    reference: PaymentReference
    status: PaymentStatus
    session_id: str | None = None
    customer_action_url: str | None = field(default=None, repr=False)
    provider_status: str | None = None
    raw: dict[str, Any] = field(default_factory=dict, repr=False)

    @property
    def reconciliation_required(self) -> bool:
        return self.status == PaymentStatus.UNKNOWN

jukto.payments.PaymentVerificationResult dataclass

Source code in src/jukto/payments.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
@dataclass(frozen=True)
class PaymentVerificationResult:
    reference: PaymentReference
    status: PaymentStatus
    validation_id: str | None = None
    bank_transaction_id: str | None = None
    requires_review: bool = True
    provider_status: str | None = None
    raw: dict[str, Any] = field(default_factory=dict, repr=False)

    @property
    def can_fulfill(self) -> bool:
        return self.status == PaymentStatus.PAID and not self.requires_review

    @property
    def reconciliation_required(self) -> bool:
        return self.status in (PaymentStatus.UNKNOWN, PaymentStatus.PENDING)

jukto.payments.PaymentQueryResult dataclass

Source code in src/jukto/payments.py 
148
149
150
151
152
@dataclass(frozen=True)
class PaymentQueryResult:
    reference: PaymentReference
    attempts: tuple[PaymentVerificationResult, ...]
    raw: dict[str, Any] = field(default_factory=dict, repr=False)

jukto.payments.RefundRequest dataclass

Source code in src/jukto/payments.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
@dataclass(frozen=True)
class RefundRequest:
    payment: PaymentReference
    bank_transaction_id: str
    merchant_refund_id: str
    amount: Decimal
    remarks: str

    def __post_init__(self) -> None:
        text(self.bank_transaction_id, "bank_transaction_id", 80)
        text(self.merchant_refund_id, "merchant_refund_id", 30)
        text(self.remarks, "remarks", 255)
        amount(self.amount)
        if self.amount > self.payment.amount:
            raise JuktoValidationError(
                "Refund exceeds the original payment amount", provider="SSLCommerz"
            )

jukto.payments.RefundStatus

Bases: str, Enum

Source code in src/jukto/payments.py 
21
22
23
24
25
class RefundStatus(str, Enum):
    PENDING = "pending"
    REFUNDED = "refunded"
    FAILED = "failed"
    UNKNOWN = "unknown"

jukto.payments.RefundResult dataclass

Source code in src/jukto/payments.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
@dataclass(frozen=True)
class RefundResult:
    bank_transaction_id: str
    status: RefundStatus
    provider_refund_id: str | None
    merchant_refund_id: str | None = None
    requested_amount: Decimal | None = None
    provider_status: str | None = None
    raw: dict[str, Any] = field(default_factory=dict, repr=False)
    currency: str | None = None
    merchant_transaction_id: str | None = None

    @property
    def reconciliation_required(self) -> bool:
        return self.status in (RefundStatus.PENDING, RefundStatus.UNKNOWN)

jukto.payments.PaymentVerificationError

Bases: ProviderError

Notification or authenticated response does not match trusted payment data.

Source code in src/jukto/payments.py 
28
29
class PaymentVerificationError(ProviderError):
    """Notification or authenticated response does not match trusted payment data."""