Initial release: qbirr v1.0.0

src/qbirr/__init__.py

@@ -0,0 +1,192 @@
+"""
+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
+        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