"""Payment gateway abstraction. Backends implement two operations: start(order, callback_url) -> redirect_url Open a payment intent on the gateway and return the URL to redirect the user's browser to. May raise PaymentGatewayError. verify(order) -> (ok: bool, ref_id: str | None, error: str | None) Confirm the payment after the user has been bounced back to our callback. The Order's `authority` field must already be populated. Two backends are bundled: ConsolePaymentBackend Dev-only. `start()` returns a URL pointing back to our own callback view with success query parameters, so the dev can click through and complete the flow without a real gateway. `verify()` always succeeds. ZarinPalBackend Production. Talks to ZarinPal v4 (sandbox or live, toggled by ZARINPAL_SANDBOX). Lifted from the working Vitron-back implementation with the cleanups identified in the design review (no hardcoded sandbox URLs, explicit timeouts, no over-logging of merchant id / payloads). Selection is via the dotted path in `settings.PAYMENT_BACKEND`. """ from __future__ import annotations import logging from typing import Optional, Protocol, Tuple import requests from django.conf import settings from django.utils.module_loading import import_string logger = logging.getLogger(__name__) # ----------------------------------------------------------------------------- exceptions class PaymentGatewayError(Exception): """Raised when a gateway call fails or returns an unexpected response.""" # ----------------------------------------------------------------------------- protocol class PaymentBackend(Protocol): gateway_id: str def start(self, order, callback_url: str) -> str: ... def verify(self, order) -> Tuple[bool, Optional[str], Optional[str]]: ... # ----------------------------------------------------------------------------- console (dev) class ConsolePaymentBackend: """Dev backend — fakes the gateway round-trip. `start()` returns the callback URL with a success token, so the user clicks a link and lands on our callback view as if they'd successfully paid. """ gateway_id = "console" def start(self, order, callback_url: str) -> str: # Embed a fake authority + Status=OK so the callback succeeds when the # browser is pointed here. The authority is what we'll match against # when the callback runs, so we also store it on the order. authority = f"CONSOLE-{order.pk}" order.authority = authority order.save(update_fields=["authority"]) sep = "&" if "?" in callback_url else "?" return f"{callback_url}{sep}Authority={authority}&Status=OK" def verify(self, order) -> Tuple[bool, Optional[str], Optional[str]]: # Always succeed in dev. Pretend ZarinPal gave us a ref_id. return True, f"console-ref-{order.pk}", None # ----------------------------------------------------------------------------- ZarinPal class ZarinPalBackend: """Hand-rolled ZarinPal v4 client. Endpoints: Request: POST {base}/pg/v4/payment/request.json Verify: POST {base}/pg/v4/payment/verify.json Redirect: {pg_base}/pg/StartPay/ Amounts are stored in **Toman** in our DB; ZarinPal expects **Rial**, so we multiply by 10 on every API call. """ gateway_id = "zarinpal" TIMEOUT = 15 # API endpoints (sandbox vs production) SANDBOX_API_BASE = "https://sandbox.zarinpal.com" LIVE_API_BASE = "https://api.zarinpal.com" SANDBOX_PG_BASE = "https://sandbox.zarinpal.com" LIVE_PG_BASE = "https://www.zarinpal.com" def __init__(self): self.merchant_id = settings.ZARINPAL_MERCHANT_ID self.sandbox = bool(getattr(settings, "ZARINPAL_SANDBOX", True)) if not self.merchant_id: raise PaymentGatewayError("ZARINPAL_MERCHANT_ID is not configured") # -- internal helpers --------------------------------------------------- @property def _api_base(self) -> str: return self.SANDBOX_API_BASE if self.sandbox else self.LIVE_API_BASE @property def _pg_base(self) -> str: return self.SANDBOX_PG_BASE if self.sandbox else self.LIVE_PG_BASE def _post(self, path: str, payload: dict) -> dict: url = f"{self._api_base}{path}" try: resp = requests.post(url, json=payload, timeout=self.TIMEOUT) except requests.RequestException as e: raise PaymentGatewayError(f"ZarinPal network error: {e}") from e try: data = resp.json() except ValueError as e: raise PaymentGatewayError( f"ZarinPal non-JSON response (HTTP {resp.status_code})" ) from e if not (200 <= resp.status_code < 300): raise PaymentGatewayError(f"ZarinPal HTTP {resp.status_code}: {data}") return data # -- protocol ----------------------------------------------------------- def start(self, order, callback_url: str) -> str: amount_rial = int(order.amount_toman) * 10 payload = { "merchant_id": self.merchant_id, "amount": amount_rial, "callback_url": callback_url, "description": f"ilo · {order.course.title or 'course'} (#{order.pk})", "metadata": { "mobile": getattr(order.student, "phone_number", "") or "", }, } data = self._post("/pg/v4/payment/request.json", payload) body = data.get("data") or {} if data.get("errors") or body.get("code") != 100 or not body.get("authority"): errors = data.get("errors") or body raise PaymentGatewayError(f"ZarinPal request failed: {errors}") authority = body["authority"] order.authority = authority order.save(update_fields=["authority"]) return f"{self._pg_base}/pg/StartPay/{authority}" def verify(self, order) -> Tuple[bool, Optional[str], Optional[str]]: if not order.authority: return False, None, "Order has no authority to verify" amount_rial = int(order.amount_toman) * 10 payload = { "merchant_id": self.merchant_id, "amount": amount_rial, "authority": order.authority, } try: data = self._post("/pg/v4/payment/verify.json", payload) except PaymentGatewayError as e: return False, None, str(e) body = data.get("data") or {} code = body.get("code") # 100 = first-time success, 101 = already verified (treat as success) if code in (100, 101): return True, str(body.get("ref_id") or ""), None errors = data.get("errors") or body return False, None, f"ZarinPal verify failed: {errors}" # ----------------------------------------------------------------------------- IDPay class IDPayBackend: """Hand-rolled IDPay v1.1 client. Endpoints: Create: POST {base}/v1.1/payment body: {order_id, amount, name, phone, desc, callback} Verify: POST {base}/v1.1/payment/verify body: {id, order_id} Redirect: response.link (returned by Create) Like ZarinPal, amounts go in **Rial** (Toman * 10). Sandbox toggled via the `X-SANDBOX: 1` header rather than a separate URL. """ gateway_id = "idpay" TIMEOUT = 15 BASE_URL = "https://api.idpay.ir" def __init__(self): self.api_key = settings.IDPAY_API_KEY self.sandbox = bool(getattr(settings, "IDPAY_SANDBOX", True)) if not self.api_key: raise PaymentGatewayError("IDPAY_API_KEY is not configured") def _headers(self): h = {"X-API-KEY": self.api_key, "Content-Type": "application/json"} if self.sandbox: h["X-SANDBOX"] = "1" return h def _post(self, path: str, payload: dict) -> dict: url = f"{self.BASE_URL}{path}" try: resp = requests.post(url, json=payload, headers=self._headers(), timeout=self.TIMEOUT) except requests.RequestException as e: raise PaymentGatewayError(f"IDPay network error: {e}") from e try: data = resp.json() except ValueError as e: raise PaymentGatewayError(f"IDPay non-JSON response (HTTP {resp.status_code})") from e if not (200 <= resp.status_code < 300): raise PaymentGatewayError(f"IDPay HTTP {resp.status_code}: {data}") return data def start(self, order, callback_url: str) -> str: amount_rial = int(order.amount_toman) * 10 payload = { "order_id": str(order.pk), "amount": amount_rial, "name": order.student.full_name or "", "phone": order.student.phone_number or "", "desc": f"ilo · {order.course.title or 'course'} (#{order.pk})", "callback": callback_url, } data = self._post("/v1.1/payment", payload) if not data.get("id") or not data.get("link"): raise PaymentGatewayError(f"IDPay request failed: {data}") order.authority = data["id"] order.save(update_fields=["authority"]) return data["link"] def verify(self, order) -> Tuple[bool, Optional[str], Optional[str]]: if not order.authority: return False, None, "Order has no authority to verify" payload = {"id": order.authority, "order_id": str(order.pk)} try: data = self._post("/v1.1/payment/verify", payload) except PaymentGatewayError as e: return False, None, str(e) status_code = data.get("status") # 100 = first-time success, 101 = already verified if status_code in (100, 101): return True, str(data.get("track_id") or ""), None return False, None, f"IDPay verify failed: {data}" # ----------------------------------------------------------------------------- factory _DEFAULT_BACKEND = "apps.payments.services.gateway.ConsolePaymentBackend" def get_payment_backend() -> PaymentBackend: return import_string(getattr(settings, "PAYMENT_BACKEND", _DEFAULT_BACKEND))()