Python SDK

مرجع كامل لـ l402kit Python SDK — مزخرف FastAPI وFlask لواجهات برمجة التطبيقات المدفوعة عبر Bitcoin Lightning.

التثبيت

pip install l402kit

المتطلبات: Python 3.11+، FastAPI أو Flask (اختياري)

---

وضع Soberano (تحتفظ بـ 100%)

أحضر مزود Lightning الخاص بك — تذهب المدفوعات مباشرةً إلى محفظتك، بدون رسوم.

import os
from fastapi import FastAPI, Request
from l402kit import l402_required
from l402kit.providers.blink import BlinkProvider

app = FastAPI()

lightning = BlinkProvider(
    api_key=os.environ["BLINK_API_KEY"],
    wallet_id=os.environ["BLINK_WALLET_ID"],
)

@app.get("/api/data")
@l402_required(price_sats=10, lightning=lightning)
async def get_data(request: Request):
    return {"data": "premium content"}

يدعم Python أيضًا وضع Managed — استخدم ManagedProvider.from_address("[e-mail]") (رسوم 0.3%، لا حاجة لعقدة). راجع قسم المزودين أدناه.

---

Flask

import os
from flask import Flask, jsonify
from l402kit import l402_required
from l402kit.providers.blink import BlinkProvider

app = Flask(__name__)

lightning = BlinkProvider(
    api_key=os.environ["BLINK_API_KEY"],
    wallet_id=os.environ["BLINK_WALLET_ID"],
)

@app.route("/api/data")
@l402_required(price_sats=10, lightning=lightning)
def get_data():
    return jsonify({"data": "premium content"})

if __name__ == "__main__":
    app.run(port=3000)

Flask + Gunicorn مع عمال gevent/eventlet: يستدعي l402_required مزود Lightning غير المتزامن من معالج Flask متزامن. إذا كان عامل Gunicorn يستخدم monkey-patching لـ gevent أو eventlet (مما يُنشئ حلقة أحداث قيد التشغيل)، يكتشف المزخرف ذلك تلقائيًا وينفّذ الاستدعاء غير المتزامن في خيط مخصص — لا يلزم اتخاذ أي إجراء. عمال Gunicorn المتزامنون القياسيون وuvicorn (FastAPI) غير متأثرين.

---

@l402_required — المزخرف

المعاملات

المعامل	النوع	الافتراضي	الوصف
price_sats	int	مطلوب	السعر لكل استدعاء بالـ satoshis
lightning	LightningProvider	مطلوب	خلفية Lightning الخاصة بك
replay	ReplayAdapter	في الذاكرة	خلفية حماية قابلة للتوصيل من إعادة التشغيل

السلوك

الطلب	الاستجابة
لا يوجد رأس Authorization	402 مع فاتورة، macaroon، سعر
L402 <macaroon>:<preimage> صالح	يُنفَّذ المعالج بشكل طبيعي
رمز غير صالح أو منتهي الصلاحية	401 Unauthorized
preimage مُعاد استخدامه	401 Token already used

استجابة 402

{
  "error": "Payment Required",
  "price_sats": 10,
  "invoice": "lnbc100n1...",
  "macaroon": "eyJoYXNoIjoiYWJjMTIzIiwiZXhwIjoxNzAwMDAwMDAwfQ=="
}

---

المزودون

BlinkProvider

Blink — محفظة Lightning وصائية مجانية، لا KYC للمبالغ الصغيرة.

from l402kit.providers.blink import BlinkProvider

blink = BlinkProvider(
    api_key=os.environ["BLINK_API_KEY"],    # dashboard.blink.sv → API Keys
    wallet_id=os.environ["BLINK_WALLET_ID"],
)

LNbitsProvider

from l402kit.providers.lnbits import LNbitsProvider

lnbits = LNbitsProvider(
    api_key=os.environ["LNBITS_API_KEY"],
    base_url="https://your-lnbits.com",  # optional
)

OpenNodeProvider

from l402kit.providers.opennode import OpenNodeProvider

opennode = OpenNodeProvider(
    api_key=os.environ["OPENNODE_API_KEY"],
    test_mode=False,  # True for sandbox
)

ManagedProvider — وضع السحابة (رسوم 0.3%)

تستضيف l402kit.com عقدة Lightning. تستلم 99.7% من كل دفعة — لا يلزم إعداد عقدة.

from l402kit import ManagedProvider

lightning = ManagedProvider.from_address("[e-mail]")

# Optional: register in the public API directory
lightning = ManagedProvider.from_address("[e-mail]", register_directory={
    "url": "https://api.you.com/v1/weather",
    "name": "Weather API",
    "price_sats": 10,
    "category": "weather",
})

---

حماية إعادة التشغيل

الافتراضي — في الذاكرة (التطوير)

مدمج، لا يلزم أي إعداد. يُعاد ضبطه عند إعادة تشغيل العملية.

Redis (الإنتاج — متعدد النسخ)

لنشر Gunicorn/uvicorn متعدد العمال، شارك حالة إعادة التشغيل عبر Redis:

import os, redis
from l402kit import l402_required, RedisReplayAdapter

r = redis.Redis.from_url(os.environ["REDIS_URL"])
replay = RedisReplayAdapter(r, ttl_seconds=86400)

@app.get("/api/data")
@l402_required(
    price_sats=10,
    lightning=lightning,
    replay=replay,
)
async def get_data(request: Request):
    return {"data": "premium content"}

يستخدم RedisReplayAdapter الأمر SET key 1 NX EX ttl — ذري وخالٍ من حالات التسابق.

---

أدوات مستقلة

from l402kit.verify import verify_token
from l402kit.replay import check_and_mark_preimage

# Verify a token (True / False)
is_valid = verify_token("eyJoYXNoIjoiYWJjMTIzIiwiZXhwIjoxNzAwMDAwMDAwfQ==:deadbeef...")

# Manual replay check
is_first_use = check_and_mark_preimage(preimage)
# True = first use, False = already used

---

مزود مخصص

from l402kit.types import LightningProvider, Invoice
import base64, json, time

class MyProvider(LightningProvider):
    async def create_invoice(self, amount_sats: int) -> Invoice:
        result = await my_node.create_invoice(amount_sats)
        exp = int((time.time() + 3600) * 1000)
        macaroon = base64.b64encode(
            json.dumps({"hash": result.hash, "exp": exp}).encode()
        ).decode()
        return Invoice(
            payment_request=result.bolt11,
            payment_hash=result.hash,
            macaroon=macaroon,
            amount_sats=amount_sats,
            expires_at=exp,
        )

    async def check_payment(self, payment_hash: str) -> bool:
        return await my_node.is_paid(payment_hash)

---

الاختبار

import hashlib, base64, json, time, os
from l402kit.verify import verify_token

def make_test_token() -> str:
    preimage = os.urandom(32).hex()
    payment_hash = hashlib.sha256(bytes.fromhex(preimage)).hexdigest()
    exp = int((time.time() + 3600) * 1000)
    macaroon = base64.b64encode(
        json.dumps({"hash": payment_hash, "exp": exp}).encode()
    ).decode()
    return f"{macaroon}:{preimage}"

assert verify_token(make_test_token()) is True

---

التشغيل

# FastAPI
uvicorn main:app --port 3000

# Flask
python app.py

# Test — triggers 402
curl http://localhost:3000/api/data

# Pay invoice, then:
curl -H "Authorization: L402 <macaroon>:<preimage>" http://localhost:3000/api/data

---

L402Client — الدفع التلقائي

يُغلّف L402Client مكتبة httpx ويتولى دورة 402 → الدفع → إعادة المحاولة تلقائيًا.

from l402kit import L402Client
from l402kit.wallets import BlinkWallet

wallet = BlinkWallet(
    api_key=os.environ["BLINK_API_KEY"],
    wallet_id=os.environ["BLINK_WALLET_ID"],
)

client = L402Client(wallet=wallet)
data = client.get("https://api.example.com/premium").json()

المحافظ

الصنف	التثبيت	الوصف
BlinkWallet	pip install l402kit	الدفع عبر Blink GraphQL API
AlbyWallet	pip install l402kit	الدفع عبر Alby REST API

from l402kit.wallets import BlinkWallet, AlbyWallet

blink = BlinkWallet(api_key="...", wallet_id="...")
alby  = AlbyWallet(access_token=os.environ["ALBY_TOKEN"])

---

AsyncL402Client — async/await

يستخدم AsyncL402Client مكتبة httpx.AsyncClient داخليًا — مثالي لـ FastAPI وasyncio وأطر عمل وكلاء الذكاء الاصطناعي التي تعمل في حلقة أحداث غير متزامنة.

import asyncio
from l402kit import AsyncL402Client
from l402kit.wallets import BlinkWallet

async def main():
    async with AsyncL402Client(
        wallet=BlinkWallet(os.environ["BLINK_API_KEY"], os.environ["BLINK_WALLET_ID"]),
        budget_sats=500,
    ) as client:
        r = await client.get("https://api.example.com/premium")
        print(r.json())

asyncio.run(main())

الفرق عن L402Client

	L402Client	AsyncL402Client
عميل HTTP	httpx (متزامن)	httpx.AsyncClient
get / post	متزامن	async
الأنسب لـ	السكريبتات، Flask	FastAPI، asyncio، LangChain _arun
الميزانية / التخزين المؤقت	✅ نفسه	✅ نفسه

---

DevProvider + DevWallet — التطوير المحلي

تطوير محلي بدون إعداد — لا عقدة Lightning، ولا مدفوعات حقيقية. متطابق تشفيريًا مع الإنتاج: SHA256(preimage) === paymentHash.

from l402kit.dev import DevProvider, DevWallet
from l402kit import L402Client, l402_required
from fastapi import FastAPI, Request

app = FastAPI()
provider = DevProvider()
wallet   = DevWallet(provider)

@app.get("/premium")
@l402_required(price_sats=1, lightning=provider)
async def premium(request: Request):
    return {"data": "premium content"}

# Client — pays automatically without real Lightning
client = L402Client(wallet=wallet)
data   = client.get("http://localhost:8000/premium").json()