139 lines
5.0 KiB
Python
139 lines
5.0 KiB
Python
import os
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
import requests
|
|
|
|
|
|
class PaymentGatewayError(RuntimeError):
|
|
def __init__(self, message: str, status_code: Optional[int] = None, payload: Optional[dict] = None) -> None:
|
|
super().__init__(message)
|
|
self.status_code = status_code
|
|
self.payload = payload or {}
|
|
|
|
|
|
@dataclass
|
|
class PaymentInitResult:
|
|
external_id: str
|
|
status: Optional[str]
|
|
redirect_url: Optional[str]
|
|
payload: dict
|
|
|
|
|
|
class BasePaymentGateway:
|
|
def create_payment(
|
|
self,
|
|
amount: int,
|
|
currency: str,
|
|
description: str,
|
|
source: dict,
|
|
callback_url: Optional[str],
|
|
given_id: str,
|
|
metadata: dict,
|
|
) -> PaymentInitResult:
|
|
raise NotImplementedError
|
|
|
|
def capture_payment(self, external_id: str, amount: Optional[int] = None) -> None:
|
|
"""Capture an authorized payment. Amount in minor units; omit for full capture."""
|
|
raise NotImplementedError
|
|
|
|
def refund_payment(self, external_id: str, amount: Optional[int] = None) -> None:
|
|
"""Refund a paid/captured payment. Amount in minor units; omit for full refund."""
|
|
raise NotImplementedError
|
|
|
|
|
|
class MoyasarGateway(BasePaymentGateway):
|
|
def __init__(self) -> None:
|
|
self.secret_key = os.getenv("MOYASAR_SECRET_KEY")
|
|
self.publishable_key = os.getenv("MOYASAR_PUBLISHABLE_KEY")
|
|
self.base_url = os.getenv("MOYASAR_BASE_URL", "https://api.moyasar.com")
|
|
|
|
def _assert_config(self) -> None:
|
|
if not self.secret_key or not self.publishable_key:
|
|
raise ValueError("Moyasar credentials are not configured")
|
|
|
|
def create_payment(
|
|
self,
|
|
amount: int,
|
|
currency: str,
|
|
description: str,
|
|
source: dict,
|
|
callback_url: Optional[str],
|
|
given_id: str,
|
|
metadata: dict,
|
|
) -> PaymentInitResult:
|
|
self._assert_config()
|
|
url = f"{self.base_url}/v1/payments"
|
|
payload = {
|
|
"amount": amount,
|
|
"currency": currency,
|
|
"description": description,
|
|
"source": source,
|
|
"given_id": given_id,
|
|
"metadata": metadata,
|
|
}
|
|
if callback_url:
|
|
payload["callback_url"] = callback_url
|
|
|
|
try:
|
|
response = requests.post(url, json=payload, auth=(self.secret_key, ""), timeout=10)
|
|
except requests.RequestException as exc:
|
|
raise PaymentGatewayError("Failed to reach Moyasar") from exc
|
|
|
|
try:
|
|
data = response.json() if response.content else {}
|
|
except ValueError as exc:
|
|
raise PaymentGatewayError("Invalid response from Moyasar") from exc
|
|
|
|
if response.status_code not in (200, 201):
|
|
raise PaymentGatewayError(
|
|
"Moyasar returned an error",
|
|
status_code=response.status_code,
|
|
payload=data,
|
|
)
|
|
redirect_url = None
|
|
source_payload = data.get("source") or {}
|
|
if isinstance(source_payload, dict):
|
|
redirect_url = source_payload.get("transaction_url")
|
|
|
|
return PaymentInitResult(
|
|
external_id=data.get("id"),
|
|
status=data.get("status"),
|
|
redirect_url=redirect_url,
|
|
payload=data,
|
|
)
|
|
|
|
def capture_payment(self, external_id: str, amount: Optional[int] = None) -> None:
|
|
"""Capture an authorized payment. Amount in minor units; omit for full capture."""
|
|
self._assert_config()
|
|
url = f"{self.base_url}/v1/payments/{external_id}/capture"
|
|
payload = {} if amount is None else {"amount": amount}
|
|
try:
|
|
response = requests.post(url, json=payload, auth=(self.secret_key, ""), timeout=10)
|
|
except requests.RequestException as exc:
|
|
raise PaymentGatewayError("Failed to reach Moyasar for capture") from exc
|
|
if response.status_code not in (200, 201):
|
|
data = response.json() if response.content else {}
|
|
raise PaymentGatewayError(
|
|
"Moyasar capture failed",
|
|
status_code=response.status_code,
|
|
payload=data,
|
|
)
|
|
|
|
def refund_payment(self, external_id: str, amount: Optional[int] = None) -> None:
|
|
"""Refund a paid/captured payment. Amount in minor units; omit for full refund."""
|
|
self._assert_config()
|
|
url = f"{self.base_url}/v1/payments/{external_id}/refund"
|
|
payload = {} if amount is None else {"amount": amount}
|
|
try:
|
|
response = requests.post(url, json=payload, auth=(self.secret_key, ""), timeout=10)
|
|
except requests.RequestException as exc:
|
|
raise PaymentGatewayError("Failed to reach Moyasar for refund") from exc
|
|
if response.status_code not in (200, 201):
|
|
data = response.json() if response.content else {}
|
|
raise PaymentGatewayError(
|
|
"Moyasar refund failed",
|
|
status_code=response.status_code,
|
|
payload=data,
|
|
)
|