154 lines
4.4 KiB
Python
154 lines
4.4 KiB
Python
"""
|
|
FetcherPay API Client
|
|
"""
|
|
|
|
import hmac
|
|
import hashlib
|
|
from typing import Optional
|
|
import requests
|
|
from requests.adapters import HTTPAdapter
|
|
from urllib.parse import urljoin
|
|
|
|
from .exceptions import (
|
|
FetcherPayError,
|
|
AuthenticationError,
|
|
ValidationError,
|
|
NotFoundError,
|
|
IdempotencyError,
|
|
)
|
|
from .api.payments import PaymentsAPI
|
|
from .api.ledger import LedgerAPI
|
|
from .api.payment_methods import PaymentMethodsAPI
|
|
from .api.webhooks import WebhooksAPI
|
|
|
|
|
|
class FetcherPay:
|
|
"""
|
|
Main FetcherPay client
|
|
|
|
Args:
|
|
api_key: Your FetcherPay API key
|
|
environment: 'sandbox' or 'production'
|
|
base_url: Optional custom base URL
|
|
timeout: Request timeout in seconds (default: 30)
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
api_key: str,
|
|
environment: str = 'sandbox',
|
|
base_url: Optional[str] = None,
|
|
timeout: int = 30
|
|
):
|
|
self.api_key = api_key
|
|
self.environment = environment
|
|
self.timeout = timeout
|
|
|
|
if base_url:
|
|
self.base_url = base_url
|
|
elif environment == 'production':
|
|
self.base_url = 'https://api.fetcherpay.com/v1'
|
|
else:
|
|
self.base_url = 'https://sandbox.fetcherpay.com/v1'
|
|
|
|
# Create session with retries
|
|
self.session = requests.Session()
|
|
self.session.headers.update({
|
|
'Authorization': f'Bearer {api_key}',
|
|
'Content-Type': 'application/json',
|
|
})
|
|
|
|
# Initialize API modules
|
|
self.payments = PaymentsAPI(self)
|
|
self.ledger = LedgerAPI(self)
|
|
self.payment_methods = PaymentMethodsAPI(self)
|
|
self.webhooks = WebhooksAPI(self)
|
|
|
|
def _request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
params: Optional[dict] = None,
|
|
json: Optional[dict] = None,
|
|
headers: Optional[dict] = None,
|
|
idempotency_key: Optional[str] = None
|
|
) -> dict:
|
|
"""Make an HTTP request"""
|
|
url = urljoin(self.base_url, path)
|
|
|
|
request_headers = {}
|
|
if headers:
|
|
request_headers.update(headers)
|
|
if idempotency_key:
|
|
request_headers['Idempotency-Key'] = idempotency_key
|
|
|
|
try:
|
|
response = self.session.request(
|
|
method=method,
|
|
url=url,
|
|
params=params,
|
|
json=json,
|
|
headers=request_headers,
|
|
timeout=self.timeout
|
|
)
|
|
|
|
# Handle errors
|
|
if response.status_code >= 400:
|
|
self._handle_error(response)
|
|
|
|
return response.json() if response.content else {}
|
|
|
|
except requests.exceptions.Timeout:
|
|
raise FetcherPayError('Request timeout', 'timeout_error')
|
|
except requests.exceptions.ConnectionError:
|
|
raise FetcherPayError('Connection error', 'connection_error')
|
|
|
|
def _handle_error(self, response):
|
|
"""Handle API errors"""
|
|
try:
|
|
data = response.json()
|
|
error = data.get('error', {})
|
|
except:
|
|
error = {}
|
|
|
|
message = error.get('message', 'An error occurred')
|
|
error_type = error.get('type', 'api_error')
|
|
param = error.get('param')
|
|
code = error.get('code')
|
|
|
|
if response.status_code == 401:
|
|
raise AuthenticationError(message)
|
|
elif response.status_code == 404:
|
|
raise NotFoundError(message)
|
|
elif response.status_code == 422:
|
|
raise ValidationError(message, param)
|
|
elif response.status_code == 409:
|
|
raise IdempotencyError(message)
|
|
else:
|
|
raise FetcherPayError(message, error_type, response.status_code, param, code)
|
|
|
|
def verify_webhook_signature(
|
|
self,
|
|
payload: str,
|
|
signature: str,
|
|
secret: str
|
|
) -> bool:
|
|
"""
|
|
Verify a webhook signature
|
|
|
|
Args:
|
|
payload: The raw webhook payload string
|
|
signature: The signature from X-FetcherPay-Signature header
|
|
secret: Your webhook secret
|
|
|
|
Returns:
|
|
True if signature is valid
|
|
"""
|
|
expected = hmac.new(
|
|
secret.encode('utf-8'),
|
|
payload.encode('utf-8'),
|
|
hashlib.sha256
|
|
).hexdigest()
|
|
|
|
return hmac.compare_digest(signature, expected)
|