Aller au contenu

Référence — kkiapay.py

Implémentation spécifique pour le provider Kkiapay.

Logic

paygate_africa.kkiapay.client.KkiapayProvider

Bases: PaymentProvider

Payment provider implementation for Kkiapay.

Source code in src/paygate_africa/kkiapay/client.py
class KkiapayProvider(PaymentProvider):
    """
    Payment provider implementation for Kkiapay.
    """

    @property
    def conf(self):
        """
        Returns the Kkiapay configuration settings.
        """
        return get_kkiapay_settings()

    async def initiate_payment(self, tx: Transaction) -> str:
        """
        Return the render URL for Kkiapay.
        Kkiapay usually requires a frontend widget; this URL directs to the widget renderer.
        """
        return self.conf.KKIAPAY_RENDER_URL_TEMPLATE.format(id=tx.id)

    async def verify_payment(self, transaction_id: str) -> dict[str, Any]:
        """
        Verify the status of a transaction with the Kkiapay API.
        """
        base_url = self.conf.KKIAPAY_SANDBOX_URL if self.conf.KKIAPAY_SANDBOX else self.conf.KKIAPAY_URL
        url = urljoin(base_url, self.conf.KKIAPAY_TRANSACTION_STATUS_URL)
        headers = {
            "X-SECRET-KEY": self.conf.KKIAPAY_SECRET_KEY,
            "X-API-KEY": self.conf.KKIAPAY_PUBLIC_KEY,
            "X-PRIVATE-KEY": self.conf.KKIAPAY_PRIVATE_KEY,
        }
        payload = {"transactionId": transaction_id}

        try:
            data = await asyncio.to_thread(post_json, url, payload, headers)
        except urllib.error.HTTPError as exc:
            body = exc.read().decode("utf-8", errors="replace")
            raise RuntimeError(f"Kkiapay API Error: {body}") from exc

        raw_status = data.get("status", "").upper()
        if raw_status in ("SUCCESS", "COMPLETE", "COMPLETED"):
            status = "SUCCESS"
        elif raw_status in ("PENDING", "INITIATED"):
            status = "PENDING"
        else:
            status = "FAILED"

        return {
            "status": status,
            "raw_data": data,
        }

conf property

Returns the Kkiapay configuration settings.

initiate_payment(tx) async

Return the render URL for Kkiapay. Kkiapay usually requires a frontend widget; this URL directs to the widget renderer.

Source code in src/paygate_africa/kkiapay/client.py
async def initiate_payment(self, tx: Transaction) -> str:
    """
    Return the render URL for Kkiapay.
    Kkiapay usually requires a frontend widget; this URL directs to the widget renderer.
    """
    return self.conf.KKIAPAY_RENDER_URL_TEMPLATE.format(id=tx.id)

verify_payment(transaction_id) async

Verify the status of a transaction with the Kkiapay API.

Source code in src/paygate_africa/kkiapay/client.py
async def verify_payment(self, transaction_id: str) -> dict[str, Any]:
    """
    Verify the status of a transaction with the Kkiapay API.
    """
    base_url = self.conf.KKIAPAY_SANDBOX_URL if self.conf.KKIAPAY_SANDBOX else self.conf.KKIAPAY_URL
    url = urljoin(base_url, self.conf.KKIAPAY_TRANSACTION_STATUS_URL)
    headers = {
        "X-SECRET-KEY": self.conf.KKIAPAY_SECRET_KEY,
        "X-API-KEY": self.conf.KKIAPAY_PUBLIC_KEY,
        "X-PRIVATE-KEY": self.conf.KKIAPAY_PRIVATE_KEY,
    }
    payload = {"transactionId": transaction_id}

    try:
        data = await asyncio.to_thread(post_json, url, payload, headers)
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"Kkiapay API Error: {body}") from exc

    raw_status = data.get("status", "").upper()
    if raw_status in ("SUCCESS", "COMPLETE", "COMPLETED"):
        status = "SUCCESS"
    elif raw_status in ("PENDING", "INITIATED"):
        status = "PENDING"
    else:
        status = "FAILED"

    return {
        "status": status,
        "raw_data": data,
    }

Configuration

paygate_africa.kkiapay.settings.KkiapaySettings

Configuration settings for Kkiapay.

Source code in src/paygate_africa/kkiapay/settings.py
class KkiapaySettings:
    """
    Configuration settings for Kkiapay.
    """

    def __init__(self):
        self.KKIAPAY_PUBLIC_KEY: str = require_env("KKIAPAY_PUBLIC_KEY")
        self.KKIAPAY_PRIVATE_KEY: str = require_env("KKIAPAY_PRIVATE_KEY")
        self.KKIAPAY_SECRET_KEY: str = require_env("KKIAPAY_SECRET_KEY")
        self.KKIAPAY_URL: str = os.environ.get("KKIAPAY_URL", "https://api.kkiapay.me")
        self.KKIAPAY_SANDBOX_URL: str = os.environ.get(
            "KKIAPAY_SANDBOX_URL", "https://api-sandbox.kkiapay.me"
        )
        self.KKIAPAY_TRANSACTION_STATUS_URL: str = os.environ.get(
            "KKIAPAY_TRANSACTION_STATUS_URL", "/api/v1/transactions/status"
        )
        self.KKIAPAY_SANDBOX: bool = os.environ.get(
            "KKIAPAY_SANDBOX", "true"
        ).lower() in ("true", "1", "yes")
        self.KKIAPAY_RENDER_URL_TEMPLATE: str = os.environ.get(
            "KKIAPAY_RENDER_URL_TEMPLATE", "/payment/render/{id}"
        )

paygate_africa.kkiapay.settings.get_kkiapay_settings() cached

Returns a cached instance of KkiapaySettings.

Source code in src/paygate_africa/kkiapay/settings.py
@lru_cache
def get_kkiapay_settings() -> KkiapaySettings:
    """
    Returns a cached instance of KkiapaySettings.
    """
    return KkiapaySettings()