इंस्टॉलेशन
आवश्यकताएं: Python 3.11+, FastAPI या Flask (वैकल्पिक)
Soberano मोड (आप 100% रखते हैं)
अपना Lightning provider लाएं — भुगतान सीधे आपके wallet में जाता है, 0% शुल्क।
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("you@blink.sv") उपयोग करें (0.3% शुल्क, कोई नोड की जरूरत नहीं)। नीचे Providers अनुभाग देखें।
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 workers: l402_required एक sync Flask handler से async Lightning provider को कॉल करता है। यदि आपका Gunicorn worker gevent या eventlet monkey-patching उपयोग करता है (जो एक running event loop बनाता है), तो decorator इसे स्वचालित रूप से पहचानता है और async कॉल को एक dedicated thread में चलाता है — कोई कार्रवाई आवश्यक नहीं। Standard Gunicorn sync workers और uvicorn (FastAPI) अप्रभावित रहते हैं।
@l402_required — decorator
पैरामीटर
| पैरामीटर | प्रकार | डिफ़ॉल्ट | विवरण |
|---|
price_sats | int | आवश्यक | प्रति कॉल satoshis में मूल्य |
lightning | LightningProvider | आवश्यक | आपका Lightning backend |
replay | ReplayAdapter | in-memory | Pluggable replay protection backend |
व्यवहार
| अनुरोध | प्रतिक्रिया |
|---|
कोई Authorization हेडर नहीं | invoice, macaroon, मूल्य के साथ 402 |
वैध L402 <macaroon>:<preimage> | Handler सामान्य रूप से execute होता है |
| अमान्य या expired token | 401 Unauthorized |
| Replayed preimage | 401 Token already used |
402 प्रतिक्रिया
{
"error": "Payment Required",
"price_sats": 10,
"invoice": "lnbc100n1...",
"macaroon": "eyJoYXNoIjoiYWJjMTIzIiwiZXhwIjoxNzAwMDAwMDAwfQ=="
}
Providers
BlinkProvider
Blink — मुफ्त custodial Lightning wallet, छोटी राशियों के लिए कोई 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 — cloud मोड (0.3% शुल्क)
l402kit.com Lightning नोड होस्ट करता है। आप प्रत्येक भुगतान का 99.7% प्राप्त करते हैं — कोई नोड सेटअप आवश्यक नहीं।
from l402kit import ManagedProvider
lightning = ManagedProvider.from_address("you@blink.sv")
# वैकल्पिक: सार्वजनिक API directory में पंजीकृत करें
lightning = ManagedProvider.from_address("you@blink.sv", register_directory={
"url": "https://api.you.com/v1/weather",
"name": "Weather API",
"price_sats": 10,
"category": "weather",
})
Replay protection
डिफ़ॉल्ट — in-memory (विकास)
Built-in, कोई कॉन्फ़िगरेशन आवश्यक नहीं। Process restart पर रीसेट होता है।
Redis (production — multi-instance)
Gunicorn/uvicorn multi-worker deployments के लिए, Redis के माध्यम से replay state साझा करें:
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 उपयोग करता है — atomic और race-condition मुक्त।
Standalone utilities
from l402kit.verify import verify_token
from l402kit.replay import check_and_mark_preimage
# एक token verify करें (True / False)
is_valid = verify_token("eyJoYXNoIjoiYWJjMTIzIiwiZXhwIjoxNzAwMDAwMDAwfQ==:deadbeef...")
# Manual replay जांच
is_first_use = check_and_mark_preimage(preimage)
# True = पहला उपयोग, False = पहले ही उपयोग किया जा चुका है
Custom provider
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
# परीक्षण — 402 ट्रिगर करता है
curl http://localhost:3000/api/data
# invoice का भुगतान करें, फिर:
curl -H "Authorization: L402 <macaroon>:<preimage>" http://localhost:3000/api/data
L402Client — स्वचालित भुगतान
L402Client httpx को wrap करता है और पूरे 402 → pay → retry loop को स्वचालित रूप से संभालता है।
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()
Wallets
| Class | इंस्टॉल | विवरण |
|---|
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, और AI agent frameworks के लिए आदर्श जो async event loop में चलते हैं।
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 client | httpx (sync) | httpx.AsyncClient |
get / post | sync | async |
| सबसे उपयुक्त | Scripts, Flask | FastAPI, asyncio, LangChain _arun |
| Budget / cache | ✅ समान | ✅ समान |
DevProvider + DevWallet — स्थानीय विकास
Zero-config स्थानीय विकास — कोई Lightning नोड नहीं, कोई वास्तविक भुगतान नहीं। Cryptographically production के समान: 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 — वास्तविक Lightning के बिना स्वचालित रूप से भुगतान करता है
client = L402Client(wallet=wallet)
data = client.get("http://localhost:8000/premium").json()
