Skip to content

Providers

Base Class and Registry

Provider

Bases: ABC

Abstract base class for payment provider integrations.

Source code in merchants/providers/__init__.py
class Provider(ABC):
    """Abstract base class for payment provider integrations."""

    #: Short machine-readable identifier, e.g. ``"stripe"``.
    key: str = "base"
    #: Human-readable provider name, e.g. ``"Stripe"``.
    name: str = "Base"
    #: Author or maintainer of this provider integration.
    author: str = ""
    #: Version string for this provider integration.
    version: str = "0.0.0"
    #: Short description of the provider.
    description: str = ""
    #: Documentation or homepage URL.
    url: str = ""
    #: Whether this provider accepts a webhook notification URL.
    #: Subclasses that support server-to-server webhook callbacks should
    #: override this to ``True`` so that :meth:`PaymentMixin.create` will
    #: automatically inject the calculated ``notify_url`` into the provider
    #: call.  Internal providers (e.g. ``saldo``, ``cafeteria``) should
    #: leave this as ``False``.
    accepts_notify_url: bool = False

    def __init__(
        self,
        *,
        key: str | None = None,
        name: str | None = None,
        description: str | None = None,
    ) -> None:
        for attr, value in (("key", key), ("name", name), ("description", description)):
            if value is not None:
                setattr(self, attr, value)

    def get_info(self) -> ProviderInfo:
        """Return a :class:`ProviderInfo` populated from this provider's class attributes."""
        return ProviderInfo(
            key=self.key,
            name=self.name,
            author=self.author,
            version=self.version,
            description=self.description,
            url=self.url,
        )

    @abstractmethod
    def create_checkout(
        self,
        amount: Decimal,
        currency: str,
        success_url: str,
        cancel_url: str,
        metadata: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> CheckoutSession:
        """Create a hosted-checkout session.

        Args:
            kwargs: Provider-specific keyword arguments (e.g. ``notify_url``).

        Returns:
            :class:`~merchants.models.CheckoutSession` with a ``redirect_url``.

        Raises:
            :class:`UserError`: If the provider returns an error response.
        """

    @abstractmethod
    def get_payment(self, payment_id: str) -> PaymentStatus:
        """Retrieve and normalise the status of a payment.

        Returns:
            :class:`~merchants.models.PaymentStatus` with a normalised
            :class:`~merchants.models.PaymentState`.
        """

    @abstractmethod
    def parse_webhook(self, payload: bytes, headers: dict[str, str]) -> WebhookEvent:
        """Parse and normalise a raw webhook payload.

        Returns:
            :class:`~merchants.models.WebhookEvent`.
        """

Functions

get_info

get_info() -> ProviderInfo

Return a :class:ProviderInfo populated from this provider's class attributes.

Source code in merchants/providers/__init__.py
def get_info(self) -> ProviderInfo:
    """Return a :class:`ProviderInfo` populated from this provider's class attributes."""
    return ProviderInfo(
        key=self.key,
        name=self.name,
        author=self.author,
        version=self.version,
        description=self.description,
        url=self.url,
    )

create_checkout abstractmethod

create_checkout(
    amount: Decimal,
    currency: str,
    success_url: str,
    cancel_url: str,
    metadata: dict[str, Any] | None = None,
    **kwargs: Any,
) -> CheckoutSession

Create a hosted-checkout session.

Parameters:

Name Type Description Default
kwargs Any

Provider-specific keyword arguments (e.g. notify_url).

{}

Returns:

Type Description
CheckoutSession

class:~merchants.models.CheckoutSession with a redirect_url.

Raises:

Type Description

class:UserError: If the provider returns an error response.

Source code in merchants/providers/__init__.py
@abstractmethod
def create_checkout(
    self,
    amount: Decimal,
    currency: str,
    success_url: str,
    cancel_url: str,
    metadata: dict[str, Any] | None = None,
    **kwargs: Any,
) -> CheckoutSession:
    """Create a hosted-checkout session.

    Args:
        kwargs: Provider-specific keyword arguments (e.g. ``notify_url``).

    Returns:
        :class:`~merchants.models.CheckoutSession` with a ``redirect_url``.

    Raises:
        :class:`UserError`: If the provider returns an error response.
    """

get_payment abstractmethod

get_payment(payment_id: str) -> PaymentStatus

Retrieve and normalise the status of a payment.

Returns:

Type Description
PaymentStatus

class:~merchants.models.PaymentStatus with a normalised

PaymentStatus

class:~merchants.models.PaymentState.

Source code in merchants/providers/__init__.py
@abstractmethod
def get_payment(self, payment_id: str) -> PaymentStatus:
    """Retrieve and normalise the status of a payment.

    Returns:
        :class:`~merchants.models.PaymentStatus` with a normalised
        :class:`~merchants.models.PaymentState`.
    """

parse_webhook abstractmethod

parse_webhook(
    payload: bytes, headers: dict[str, str]
) -> WebhookEvent

Parse and normalise a raw webhook payload.

Returns:

Type Description
WebhookEvent

class:~merchants.models.WebhookEvent.

Source code in merchants/providers/__init__.py
@abstractmethod
def parse_webhook(self, payload: bytes, headers: dict[str, str]) -> WebhookEvent:
    """Parse and normalise a raw webhook payload.

    Returns:
        :class:`~merchants.models.WebhookEvent`.
    """

UserError

Bases: Exception

Raised when a provider returns a user-level / validation error.

Source code in merchants/providers/__init__.py
class UserError(Exception):
    """Raised when a provider returns a user-level / validation error."""

    def __init__(self, message: str, code: str | None = None) -> None:
        super().__init__(message)
        self.message = message
        self.code = code

register_provider

register_provider(provider: Provider) -> None

Register a provider instance under its :attr:~Provider.key.

Source code in merchants/providers/__init__.py
def register_provider(provider: Provider) -> None:
    """Register a provider instance under its :attr:`~Provider.key`."""
    _REGISTRY[provider.key] = provider

get_provider

get_provider(key_or_instance: str | Provider) -> Provider

Return a provider by string key or pass through a Provider instance.

Raises:

Type Description
KeyError

If key_or_instance is a string not found in the registry.

Source code in merchants/providers/__init__.py
def get_provider(key_or_instance: str | Provider) -> Provider:
    """Return a provider by string key or pass through a Provider instance.

    Raises:
        KeyError: If ``key_or_instance`` is a string not found in the registry.
    """
    if isinstance(key_or_instance, Provider):
        return key_or_instance
    try:
        return _REGISTRY[key_or_instance]
    except KeyError:
        available = list(_REGISTRY.keys())
        raise KeyError(
            f"Provider {key_or_instance!r} not registered. " f"Available: {available}"
        ) from None

list_providers

list_providers() -> list[str]

Return the keys of all registered providers.

Source code in merchants/providers/__init__.py
def list_providers() -> list[str]:
    """Return the keys of all registered providers."""
    return list(_REGISTRY.keys())

normalise_state

normalise_state(raw_state: str) -> PaymentState

Map a provider-specific status string to a :class:~merchants.models.PaymentState.

Source code in merchants/providers/__init__.py
def normalise_state(raw_state: str) -> PaymentState:
    """Map a provider-specific status string to a :class:`~merchants.models.PaymentState`."""
    return _STATE_MAP.get(raw_state.lower(), PaymentState.UNKNOWN)

Built-in Providers

StripeProvider

Bases: Provider

Stripe-like provider stub.

Demonstrates: - Converting amounts to/from minor units (cents). - Authorization: Bearer <key> auth header. - Stripe-style status strings in state normalisation.

.. note:: This is a stub - it does not call the real Stripe API. Replace base_url and inject a real transport to connect to Stripe.

Parameters:

Name Type Description Default
api_key str

Stripe secret key (sk_test_…).

required
base_url str

Override for testing; defaults to "https://api.stripe.com".

'https://api.stripe.com'
transport Transport | None

Optional custom transport.

None
Source code in merchants/providers/stripe.py
class StripeProvider(Provider):
    """Stripe-like provider stub.

    Demonstrates:
    - Converting amounts to/from minor units (cents).
    - ``Authorization: Bearer <key>`` auth header.
    - Stripe-style status strings in state normalisation.

    .. note::
        This is a stub - it does not call the real Stripe API.
        Replace ``base_url`` and inject a real transport to connect to Stripe.

    Args:
        api_key: Stripe secret key (``sk_test_…``).
        base_url: Override for testing; defaults to ``"https://api.stripe.com"``.
        transport: Optional custom transport.
    """

    key = "stripe"
    name = "Stripe"
    author = "mariofix"
    version = "2026.3.0"
    description = "Stripe payment gateway integration (stub). Converts amounts to minor units (cents)."
    url = "https://stripe.com"

    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.stripe.com",
        *,
        transport: Transport | None = None,
    ) -> None:
        self._api_key = api_key
        self._base_url = base_url.rstrip("/")
        self._transport = transport or RequestsTransport()

    def _headers(self) -> dict[str, str]:
        return {"Authorization": f"Bearer {self._api_key}"}

    def _currency_decimals(self, currency: str) -> int:
        return 0 if currency.lower() in _ZERO_DECIMAL_CURRENCIES else 2

    def create_checkout(
        self,
        amount: Decimal,
        currency: str,
        success_url: str,
        cancel_url: str,
        metadata: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> CheckoutSession:
        decimals = self._currency_decimals(currency)
        unit_amount = to_minor_units(amount, decimals=decimals)
        payload: dict[str, Any] = {
            "payment_method_types": ["card"],
            "line_items": [
                {
                    "price_data": {
                        "currency": currency.lower(),
                        "unit_amount": unit_amount,
                        "product_data": {"name": "Order"},
                    },
                    "quantity": 1,
                }
            ],
            "mode": "payment",
            "success_url": success_url,
            "cancel_url": cancel_url,
            "metadata": metadata or {},
        }
        resp = self._transport.send(
            "POST",
            f"{self._base_url}/v1/checkout/sessions",
            headers=self._headers(),
            json=payload,
        )
        if not resp.ok:
            body_msg = (
                resp.body.get("error", {}).get("message", "")
                if isinstance(resp.body, dict)
                else ""
            )
            raise UserError(
                body_msg or f"Stripe error {resp.status_code}",
                code=str(resp.status_code),
            )

        body: dict[str, Any] = resp.body if isinstance(resp.body, dict) else {}
        return CheckoutSession(
            session_id=str(body.get("id", "")),
            redirect_url=str(body.get("url", "")),
            provider=self.key,
            amount=amount,
            currency=currency,
            metadata=metadata or {},
            raw=body,
        )

    def get_payment(self, payment_id: str) -> PaymentStatus:
        resp = self._transport.send(
            "GET",
            f"{self._base_url}/v1/payment_intents/{payment_id}",
            headers=self._headers(),
        )
        body: dict[str, Any] = resp.body if isinstance(resp.body, dict) else {}
        raw_state = str(body.get("status", "unknown"))
        currency = str(body.get("currency", ""))
        amount_minor = body.get("amount")
        decimals = self._currency_decimals(currency)
        amount_decimal = (
            from_minor_units(int(amount_minor), decimals=decimals)
            if amount_minor is not None
            else None
        )
        return PaymentStatus(
            payment_id=payment_id,
            state=normalise_state(raw_state),
            provider=self.key,
            amount=amount_decimal,
            currency=currency or None,
            raw=body,
        )

    def parse_webhook(self, payload: bytes, headers: dict[str, str]) -> WebhookEvent:
        try:
            data: dict[str, Any] = json.loads(payload)
        except ValueError:
            data = {}
        event_type = str(data.get("type", "unknown"))
        obj = data.get("data", {}).get("object", {})
        raw_state = str(obj.get("status", "unknown"))
        payment_id = obj.get("id") or obj.get("payment_intent")
        return WebhookEvent(
            event_id=data.get("id"),
            event_type=event_type,
            payment_id=payment_id,
            state=normalise_state(raw_state),
            provider=self.key,
            raw=data,
        )

PayPalProvider

Bases: Provider

PayPal-like provider stub.

Demonstrates: - Sending amounts as decimal strings (e.g. "19.99"). - Authorization: Bearer <token> auth header. - PayPal-style status strings in state normalisation.

.. note:: This is a stub - it does not call the real PayPal API. Replace base_url and inject a real transport to connect to PayPal.

Parameters:

Name Type Description Default
access_token str

OAuth access token.

required
base_url str

Override for testing; defaults to "https://api-m.paypal.com".

'https://api-m.paypal.com'
transport Transport | None

Optional custom transport.

None
Source code in merchants/providers/paypal.py
class PayPalProvider(Provider):
    """PayPal-like provider stub.

    Demonstrates:
    - Sending amounts as decimal strings (e.g. ``"19.99"``).
    - ``Authorization: Bearer <token>`` auth header.
    - PayPal-style status strings in state normalisation.

    .. note::
        This is a stub - it does not call the real PayPal API.
        Replace ``base_url`` and inject a real transport to connect to PayPal.

    Args:
        access_token: OAuth access token.
        base_url: Override for testing; defaults to ``"https://api-m.paypal.com"``.
        transport: Optional custom transport.
    """

    key = "paypal"
    name = "PayPal"
    author = "mariofix"
    version = "2026.3.0"
    description = (
        "PayPal payment gateway integration (stub). Sends amounts as decimal strings."
    )
    url = "https://developer.paypal.com"

    def __init__(
        self,
        access_token: str,
        base_url: str = "https://api-m.paypal.com",
        *,
        transport: Transport | None = None,
    ) -> None:
        self._access_token = access_token
        self._base_url = base_url.rstrip("/")
        self._transport = transport or RequestsTransport()

    def _headers(self) -> dict[str, str]:
        return {
            "Authorization": f"Bearer {self._access_token}",
            "Content-Type": "application/json",
        }

    def create_checkout(
        self,
        amount: Decimal,
        currency: str,
        success_url: str,
        cancel_url: str,
        metadata: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> CheckoutSession:
        payload: dict[str, Any] = {
            "intent": "CAPTURE",
            "purchase_units": [
                {
                    "amount": {
                        "currency_code": currency.upper(),
                        "value": to_decimal_string(amount),
                    }
                }
            ],
            "application_context": {
                "return_url": success_url,
                "cancel_url": cancel_url,
            },
        }
        resp = self._transport.send(
            "POST",
            f"{self._base_url}/v2/checkout/orders",
            headers=self._headers(),
            json=payload,
        )
        if not resp.ok:
            body_msg = (
                resp.body.get("message", "") if isinstance(resp.body, dict) else ""
            )
            raise UserError(
                body_msg or f"PayPal error {resp.status_code}",
                code=str(resp.status_code),
            )

        body: dict[str, Any] = resp.body if isinstance(resp.body, dict) else {}
        redirect_url = ""
        for link in body.get("links", []):
            if link.get("rel") == "approve":
                redirect_url = link.get("href", "")
                break

        return CheckoutSession(
            session_id=str(body.get("id", "")),
            redirect_url=redirect_url,
            provider=self.key,
            amount=amount,
            currency=currency,
            metadata=metadata or {},
            raw=body,
        )

    def get_payment(self, payment_id: str) -> PaymentStatus:
        resp = self._transport.send(
            "GET",
            f"{self._base_url}/v2/checkout/orders/{payment_id}",
            headers=self._headers(),
        )
        body: dict[str, Any] = resp.body if isinstance(resp.body, dict) else {}
        raw_state = str(body.get("status", "unknown"))
        pu = body.get("purchase_units", [{}])
        amount_info = pu[0].get("amount", {}) if pu else {}
        currency = amount_info.get("currency_code")
        amount_val = amount_info.get("value")
        amount_decimal = Decimal(str(amount_val)) if amount_val is not None else None
        return PaymentStatus(
            payment_id=payment_id,
            state=normalise_state(raw_state),
            provider=self.key,
            amount=amount_decimal,
            currency=currency,
            raw=body,
        )

    def parse_webhook(self, payload: bytes, headers: dict[str, str]) -> WebhookEvent:
        try:
            data: dict[str, Any] = json.loads(payload)
        except ValueError:
            data = {}
        event_type = str(data.get("event_type", "unknown"))
        resource = data.get("resource", {})
        raw_state = str(resource.get("status", "unknown"))
        payment_id = resource.get("id")
        return WebhookEvent(
            event_id=data.get("id"),
            event_type=event_type,
            payment_id=payment_id,
            state=normalise_state(raw_state),
            provider=self.key,
            raw=data,
        )

GenericProvider

Bases: Provider

A minimal HTTP provider that POST/GET against configurable endpoints.

This is useful for custom or in-house payment gateways that follow a simple REST interface.

Parameters:

Name Type Description Default
checkout_url str

Endpoint to POST for creating a checkout session.

required
payment_url_template str

URL template with {payment_id} placeholder for fetching payment status.

required
transport Transport | None

Optional custom :class:~merchants.transport.Transport.

None
Source code in merchants/providers/generic.py
class GenericProvider(Provider):
    """A minimal HTTP provider that POST/GET against configurable endpoints.

    This is useful for custom or in-house payment gateways that follow a
    simple REST interface.

    Args:
        checkout_url: Endpoint to ``POST`` for creating a checkout session.
        payment_url_template: URL template with ``{payment_id}`` placeholder
            for fetching payment status.
        transport: Optional custom :class:`~merchants.transport.Transport`.
    """

    key = "generic"
    name = "Generic"
    author = "mariofix"
    version = "2026.3.0"
    description = (
        "Generic REST endpoint provider for custom or in-house payment gateways."
    )
    url = ""

    def __init__(
        self,
        checkout_url: str,
        payment_url_template: str,
        *,
        transport: Transport | None = None,
        extra_headers: dict[str, str] | None = None,
    ) -> None:
        self._checkout_url = checkout_url
        self._payment_url_template = payment_url_template
        self._transport = transport or RequestsTransport()
        self._extra_headers = extra_headers or {}

    def create_checkout(
        self,
        amount: Decimal,
        currency: str,
        success_url: str,
        cancel_url: str,
        metadata: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> CheckoutSession:
        payload: dict[str, Any] = {
            "amount": to_decimal_string(amount),
            "currency": currency.upper(),
            "success_url": success_url,
            "cancel_url": cancel_url,
            "metadata": metadata or {},
        }
        resp = self._transport.send(
            "POST",
            self._checkout_url,
            headers=self._extra_headers,
            json=payload,
        )
        if not resp.ok:
            raise UserError(
                f"Provider returned {resp.status_code}", code=str(resp.status_code)
            )

        body: dict[str, Any] = resp.body if isinstance(resp.body, dict) else {}
        return CheckoutSession(
            session_id=str(body.get("id", "")),
            redirect_url=str(body.get("redirect_url", "")),
            provider=self.key,
            amount=amount,
            currency=currency,
            metadata=metadata or {},
            raw=body,
        )

    def get_payment(self, payment_id: str) -> PaymentStatus:
        url = self._payment_url_template.format(payment_id=payment_id)
        resp = self._transport.send("GET", url, headers=self._extra_headers)
        body: dict[str, Any] = resp.body if isinstance(resp.body, dict) else {}
        raw_state = str(body.get("status", "unknown"))
        return PaymentStatus(
            payment_id=payment_id,
            state=normalise_state(raw_state),
            provider=self.key,
            raw=body,
        )

    def parse_webhook(self, payload: bytes, headers: dict[str, str]) -> WebhookEvent:
        import json

        try:
            data: dict[str, Any] = json.loads(payload)
        except ValueError:
            data = {}
        raw_state = str(data.get("status", "unknown"))
        return WebhookEvent(
            event_id=data.get("event_id"),
            event_type=str(data.get("event_type", "unknown")),
            payment_id=data.get("payment_id"),
            state=normalise_state(raw_state),
            provider=self.key,
            raw=data,
        )

DummyProvider

Bases: Provider

Local development / testing provider.

Returns plausible random data without hitting any real API. Useful for rapid iteration and unit-testing application code.

Parameters:

Name Type Description Default
base_url str

Fake base URL included in the redirect URL (default: "https://dummy-pay.example.com").

'https://dummy-pay.example.com'
always_state PaymentState | None

If set, every :meth:get_payment call returns this :class:~merchants.models.PaymentState instead of a random one.

None
Source code in merchants/providers/dummy.py
class DummyProvider(Provider):
    """Local development / testing provider.

    Returns plausible random data without hitting any real API.
    Useful for rapid iteration and unit-testing application code.

    Args:
        base_url: Fake base URL included in the redirect URL
            (default: ``"https://dummy-pay.example.com"``).
        always_state: If set, every :meth:`get_payment` call returns this
            :class:`~merchants.models.PaymentState` instead of a random one.
    """

    key = "dummy"
    name = "Dummy"
    author = "mariofix"
    version = "2026.3.0"
    description = "Local development provider that returns random data without calling any real API."
    url = ""

    _TERMINAL_STATES = [
        PaymentState.SUCCEEDED,
        PaymentState.FAILED,
        PaymentState.CANCELLED,
    ]

    def __init__(
        self,
        base_url: str = "https://dummy-pay.example.com",
        *,
        always_state: PaymentState | None = None,
    ) -> None:
        self._base_url = base_url.rstrip("/")
        self._always_state = always_state

    def create_checkout(
        self,
        amount: Decimal,
        currency: str,
        success_url: str,
        cancel_url: str,
        metadata: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> CheckoutSession:
        session_id = _rand_id("dummy_sess_")
        return CheckoutSession(
            session_id=session_id,
            redirect_url=f"{self._base_url}/pay/{session_id}?amount={amount}&currency={currency}",
            provider=self.key,
            amount=amount,
            currency=currency,
            metadata=metadata or {},
            raw={"simulated": True},
        )

    def get_payment(self, payment_id: str) -> PaymentStatus:
        state = self._always_state or random.choice(self._TERMINAL_STATES)
        return PaymentStatus(
            payment_id=payment_id,
            state=state,
            provider=self.key,
            raw={"simulated": True},
        )

    def parse_webhook(self, payload: bytes, headers: dict[str, str]) -> WebhookEvent:
        import json

        try:
            data: dict[str, Any] = json.loads(payload)
        except ValueError:
            data = {}
        return WebhookEvent(
            event_id=data.get("event_id", _rand_id("dummy_evt_")),
            event_type=data.get("event_type", "payment.simulated"),
            payment_id=data.get("payment_id", _rand_id("dummy_pay_")),
            state=PaymentState.SUCCEEDED,
            provider=self.key,
            raw=data,
        )