sdk-python/src/qbirr/__init__.py

"""
qbirr — Python client for the qbirr payment verification API.

    from qbirr import Qbirr
    qb = Qbirr(api_key="...")
    r = qb.verify(
        provider="cbe",
        ref="FT23001234ABC",
        amount=500,
        receiver_name="YOUR NAME",
        receiver_account="1000017692643",
    )
    print(r.verified, r.payer)
"""

from __future__ import annotations
from dataclasses import dataclass
from typing import Optional, Any, Dict
import requests

__version__ = "1.0.0"
__all__ = ["Qbirr", "QbirrError", "VerifyResult", "UsageStatus"]

DEFAULT_BASE_URL = "https://verify.qbirr.com"
DEFAULT_TIMEOUT = 30  # seconds


class QbirrError(Exception):
    """Raised on network failures, HTTP errors, or non-JSON responses."""

    def __init__(self, message: str, status: Optional[int] = None, body: Any = None):
        super().__init__(message)
        self.status = status
        self.body = body


@dataclass
class VerifyResult:
    verified: bool
    payer: str
    amount: float
    error: str


@dataclass
class UsageStatus:
    used: int
    limit: int
    remaining: int
    plan: str
    period_start: str
    period_end: str


class Qbirr:
    """Synchronous client.  For async, just wrap with run_in_executor."""

    def __init__(
        self,
        api_key: str,
        base_url: str = DEFAULT_BASE_URL,
        timeout: float = DEFAULT_TIMEOUT,
        session: Optional[requests.Session] = None,
    ):
        if not api_key:
            raise QbirrError("api_key is required")
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self._session = session or requests.Session()
        self._session.headers.update({
            "X-API-Key": api_key,
            "Accept": "application/json",
            "User-Agent": f"qbirr-python/{__version__}",
        })

    def verify(
        self,
        provider: str,
        ref: str,
        amount: float,
        receiver_name: str,
        receiver_account: Optional[str] = None,
    ) -> VerifyResult:
        """
        Verify a single transaction.

        provider: cbe | telebirr | abyssinia | dashen | mpesa | awash | ebirr | zamzam
        ref:      Transaction ref. For Awash, the SMS URL or token. For eBirr, "tenant/token" or URL.
        amount:   Expected payment in ETB.
        receiver_name:    Your receiving account's name (case-insensitive match).
        receiver_account: Required for cbe & abyssinia.
        """
        body: Dict[str, Any] = {
            "provider": provider,
            "ref": ref,
            "amount": amount,
            "receiver_name": receiver_name,
        }
        if receiver_account:
            body["receiver_account"] = receiver_account

        data = self._request("POST", "/api/v1/verify", json=body)
        return VerifyResult(
            verified=bool(data.get("verified")),
            payer=str(data.get("payer") or ""),
            amount=float(data.get("amount") or 0),
            error=str(data.get("error") or ""),
        )

    def verify_upload(
        self,
        provider: str,
        amount: float,
        receiver_name: str,
        file: bytes,
        filename: str = "upload.pdf",
        content_type: str = "application/pdf",
        receiver_account: Optional[str] = None,
    ) -> VerifyResult:
        """
        Verify by uploading a receipt file (PDF, PNG, or JPEG).

        Images with a QR code (most bank receipts have one) are decoded
        server-side and routed through the URL flow automatically.
        """
        files = {"file": (filename, file, content_type)}
        payload = {
            "provider": provider,
            "amount": str(amount),
            "receiver_name": receiver_name,
        }
        if receiver_account:
            payload["receiver_account"] = receiver_account

        try:
            resp = self._session.post(
                f"{self.base_url}/api/v1/verify/upload",
                data=payload,
                files=files,
                timeout=self.timeout,
            )
        except requests.exceptions.RequestException as e:
            raise QbirrError(f"Network error: {e}")

        try:
            data = resp.json()
        except ValueError:
            raise QbirrError(
                f"Non-JSON response (HTTP {resp.status_code})",
                status=resp.status_code, body=resp.text,
            )
        if not resp.ok:
            msg = data.get("error") if isinstance(data, dict) else f"HTTP {resp.status_code}"
            raise QbirrError(str(msg), status=resp.status_code, body=data)

        return VerifyResult(
            verified=bool(data.get("verified")),
            payer=str(data.get("payer") or ""),
            amount=float(data.get("amount") or 0),
            error=str(data.get("error") or ""),
        )

    def usage(self) -> UsageStatus:
        data = self._request("GET", "/api/v1/usage")
        return UsageStatus(
            used=int(data.get("used") or 0),
            limit=int(data.get("limit") or 0),
            remaining=int(data.get("remaining") or 0),
            plan=str(data.get("plan") or ""),
            period_start=str(data.get("period_start") or ""),
            period_end=str(data.get("period_end") or ""),
        )

    def _request(self, method: str, path: str, json: Optional[dict] = None) -> dict:
        url = f"{self.base_url}{path}"
        try:
            resp = self._session.request(method, url, json=json, timeout=self.timeout)
        except requests.exceptions.Timeout:
            raise QbirrError(f"Request timed out after {self.timeout}s")
        except requests.exceptions.RequestException as e:
            raise QbirrError(f"Network error: {e}")

        try:
            data = resp.json()
        except ValueError:
            raise QbirrError(f"Non-JSON response (HTTP {resp.status_code})", status=resp.status_code, body=resp.text)

        if not resp.ok:
            msg = data.get("error") if isinstance(data, dict) else f"HTTP {resp.status_code}"
            raise QbirrError(str(msg), status=resp.status_code, body=data)
        return data