commit 998a9bb9843a510f00112e41f29882b5bfe5ab9d Author: admin Date: Fri Jul 17 16:42:27 2026 +0200 Initial release: qbirr v1.0.0 diff --git a/README.md b/README.md new file mode 100644 index 0000000..7838a7c --- /dev/null +++ b/README.md @@ -0,0 +1,59 @@ +# qbirr (Python) + +Verify Ethiopian bank transfer receipts in real time. + +## Install + +```bash +pip install qbirr +``` + +## Quick start + +```python +from qbirr import Qbirr + +qb = Qbirr(api_key="your_key_from_dashboard") + +r = qb.verify( + provider="cbe", + ref="FT23001234ABC", + amount=500, + receiver_name="YOUR FULL NAME", + receiver_account="1000017692643", +) + +if r.verified: + print(f"✓ Paid by {r.payer}, amount: {r.amount}") +else: + print(f"✗ {r.error}") +``` + +## Providers + +| provider | Notes | +|---|---| +| `cbe` | Needs `receiver_account` | +| `telebirr` | — | +| `abyssinia` | Needs `receiver_account` | +| `dashen` | — | +| `mpesa` | — | +| `awash` | `ref` = SMS URL or its token | +| `ebirr` | `ref` = "tenant/token" or full URL | + +## Errors + +```python +from qbirr import Qbirr, QbirrError + +try: + r = qb.verify(...) +except QbirrError as e: + print(f"API error {e.status}: {e}") +``` + +Verification *failures* (wrong amount, wrong receiver) are NOT exceptions — they're returned as `VerifyResult(verified=False, error="...")`. + +## License + +MIT diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..c747fc3 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,20 @@ +[build-system] +requires = ["setuptools>=61"] +build-backend = "setuptools.build_meta" + +[project] +name = "qbirr" +version = "1.0.0" +description = "qbirr.com — Ethiopian bank verification SDK (CBE, Telebirr, Awash, Dashen, M-Pesa, Abyssinia, eBirr)" +readme = "README.md" +requires-python = ">=3.8" +license = { text = "MIT" } +keywords = ["ethiopia", "cbe", "telebirr", "awash", "dashen", "mpesa", "abyssinia", "ebirr", "payment", "verification"] +dependencies = ["requests>=2.28"] + +[project.urls] +Homepage = "https://qbirr.com" +Documentation = "https://www.qbirr.com/docs" + +[tool.setuptools.packages.find] +where = ["src"] diff --git a/src/qbirr/__init__.py b/src/qbirr/__init__.py new file mode 100644 index 0000000..e678d25 --- /dev/null +++ b/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