--- title: "Пример на Python" url: "https://docs.x-hub.online/reference/python/" description: "Готовый пример интеграции на Python." relatedDocuments: - "https://docs.x-hub.online/reference/errors/" - "https://docs.x-hub.online/reference/nodejs/" - "https://docs.x-hub.online/reference/faq/" --- # Примеры: Python Python 3.10+ с `requests` и `flask`. ```bash pip install requests flask ``` ## Клиент X-Hub ```python # xhub_client.py import os, uuid, requests BASE_URL = 'https://api.x-hub.online/api/v1' HEADERS = { 'X-Api-Key': os.environ['XHUB_API_KEY'], 'X-Api-Secret': os.environ['XHUB_API_SECRET'], } class XHubError(Exception): pass def create_payment(amount_rub, external_id, description, metadata=None): r = requests.post( f'{BASE_URL}/payments', headers={ **HEADERS, 'Idempotency-Key': str(uuid.uuid4()), 'Content-Type': 'application/json', }, json={ 'amount_rub': amount_rub, 'external_id': external_id, 'payment_method': 'sbp', 'description': description, 'metadata': metadata or {}, }, timeout=15, ) if not r.ok: raise XHubError(f'X-Hub {r.status_code}: {r.json().get("error")}') return r.json() def get_balance(): r = requests.get(f'{BASE_URL}/balance', headers=HEADERS, timeout=10) r.raise_for_status() return r.json() ``` ## Webhook receiver (Flask) ```python # webhook_server.py import os, hmac, hashlib, json from flask import Flask, request, abort app = Flask(__name__) processed = set() # в проде — Redis / БД WEBHOOK_SECRET = os.environ['XHUB_WEBHOOK_SECRET'] @app.route('/webhook/xhub', methods=['POST']) def webhook(): raw = request.get_data() sig = request.headers.get('X-Webhook-Signature', '') timestamp = request.headers.get('X-Webhook-Timestamp', '') # Подпись: HMAC-SHA256(timestamp + "." + body, secret) signature_data = timestamp.encode() + b'.' + raw expected = 'sha256=' + hmac.new( WEBHOOK_SECRET.encode(), signature_data, hashlib.sha256 ).hexdigest() if not hmac.compare_digest(sig, expected): abort(401, 'Invalid signature') event = json.loads(raw) dedupe_key = f"{event.get('payment_id') or event.get('withdrawal_id')}:{event.get('status')}" if dedupe_key in processed: return 'OK', 200 processed.add(dedupe_key) if event['event'] == 'payment.status_changed': if event['status'] == 'paid': grant_access(event['external_id'], event.get('metadata', {})) return 'OK', 200 def grant_access(external_id, metadata): print(f'Granting access for {external_id}') if __name__ == '__main__': app.run(host='0.0.0.0', port=8080) ``` ---