شرح بسيط لـ Python Requests APIs (مع مخططات وكود حقيقي)

# app/cli.py
import argparse, asyncio, json, sqlite3, sys
from . import aggregator, config

def cmd_fetch(args):
    inserted = asyncio.run(aggregator.run_once())
    print(f"Nouveaux articles : {inserted}")

def cmd_list(args):
    conn = sqlite3.connect(config.DB_PATH)
    conn.row_factory = sqlite3.Row
    
    query = "SELECT * FROM articles WHERE 1=1"
    params = []
    if args.source:
        query += " AND source = ?"
        params.append(args.source)
    if args.search:
        query += " AND (title LIKE ? OR summary LIKE ?)"
        params.extend([f"%{args.search}%"] * 2)
    query += f" ORDER BY published_at DESC LIMIT {args.limit}"
    
    for row in conn.execute(query, params):
        print(f"[{row['source']}] {row['published_at'][:10]}")
        print(f"  {row['title']}")
        print(f"  -> {row['url']}\n")

def cmd_export(args):
    conn = sqlite3.connect(config.DB_PATH)
    conn.row_factory = sqlite3.Row
    rows = [dict(r) for r in conn.execute("SELECT * FROM articles")]
    
    if args.format == "json":
        with open(args.output, "w", encoding="utf-8") as f:
            json.dump(rows, f, ensure_ascii=False, indent=2)
    elif args.format == "csv":
        import csv
        with open(args.output, "w", newline="", encoding="utf-8") as f:
            w = csv.DictWriter(f, fieldnames=rows[0].keys())
            w.writeheader()
            w.writerows(rows)
    
    print(f"{len(rows)} articles -> {args.output}")

def main():
    p = argparse.ArgumentParser(prog="news", description="Agregateur de news")
    sub = p.add_subparsers(dest="cmd", required=True)
    
    sub.add_parser("fetch", help="Recuperer les articles").set_defaults(func=cmd_fetch)
    
    pl = sub.add_parser("list", help="Lister")
    pl.add_argument("--source", help="Filtrer par source")
    pl.add_argument("--search", help="Rechercher")
    pl.add_argument("--limit", type=int, default=20)
    pl.set_defaults(func=cmd_list)
    
    pe = sub.add_parser("export", help="Exporter")
    pe.add_argument("--format", choices=["json", "csv"], default="json")
    pe.add_argument("--output", default="export.json")
    pe.set_defaults(func=cmd_export)
    
    args = p.parse_args()
    args.func(args)

if __name__ == "__main__":
    main()

python -m app.cli fetch
python -m app.cli list --limit 10
python -m app.cli list --source hackernews
python -m app.cli list --search "python"
python -m app.cli export --format csv --output news.csv

# pyproject.toml
[project.scripts]
news = "app.cli:main"

pip install -e .
news fetch
news list --limit 5

crontab -e

# كل 30 دقيقة
*/30 * * * * cd /home/user/news_agg && \
  /home/user/.venv/bin/news fetch >> logs/news.log 2>&1

# PowerShell
schtasks /Create /SC MINUTE /MO 30 /TN "NewsAgg" /TR "C:\path\to\python.exe -m app.cli fetch"

import requests, os

STRIPE_KEY = os.getenv("STRIPE_SECRET_KEY")   # sk_test_...

# يستخدم Stripe المصادقة الأساسية مع المفتاح كمستخدم
r = requests.post(
    "https://api.stripe.com/v1/payment_intents",
    auth=(STRIPE_KEY, ""),
    data={
        "amount": 2000,           # 20.00 EUR
        "currency": "eur",
        "payment_method_types[]": "card",
        "description": "Commande #1234"
    },
    timeout=10
)
r.raise_for_status()
intent = r.json()
print(f"client_secret : {intent['client_secret']}")
# يستخدم الواجهة الأمامية client_secret لإكمال الدفع

r = requests.post("https://api.stripe.com/v1/customers",
                   auth=(STRIPE_KEY, ""),
                   data={"email": "alice@example.com",
                         "name": "Alice Dupont"})
customer = r.json()
print(customer["id"])   # cus_xxx

pip install stripe

import stripe
stripe.api_key = STRIPE_KEY

# أكثر بايثونية
intent = stripe.PaymentIntent.create(
    amount=2000,
    currency="eur",
    payment_method_types=["card"],
    description="Commande #1234"
)
print(intent.id)
print(intent.client_secret)

SENDGRID_KEY = os.getenv("SENDGRID_API_KEY")

payload = {
    "personalizations": [{
        "to": [{"email": "destinataire@example.com"}],
        "subject": "Hello depuis Python"
    }],
    "from": {"email": "contact@mondomaine.com", "name": "Mon App"},
    "content": [{
        "type": "text/html",
        "value": "<h1>Bonjour !</h1><p>Envoye via API.</p>"
    }]
}

r = requests.post(
    "https://api.sendgrid.com/v3/mail/send",
    json=payload,
    headers={"Authorization": f"Bearer {SENDGRID_KEY}"},
    timeout=10
)
print(r.status_code)   # 202 Accepted

payload = {
    "personalizations": [{
        "to": [{"email": "alice@example.com"}],
        "dynamic_template_data": {
            "name": "Alice",
            "order_id": "#1234",
            "total": "29.99 EUR"
        }
    }],
    "from": {"email": "orders@mondomaine.com"},
    "template_id": "d-xxxxx"     # تم إنشاؤه في SendGrid
}
requests.post("https://api.sendgrid.com/v3/mail/send",
               json=payload,
               headers={"Authorization": f"Bearer {SENDGRID_KEY}"})

import uuid

r = requests.post(
    "https://api.stripe.com/v1/charges",
    auth=(STRIPE_KEY, ""),
    headers={"Idempotency-Key": str(uuid.uuid4())},
    data={...}
)
# في حالة إعادة المحاولة، نفس المفتاح = لا فوترة مزدوجة

import requests

# requests يقوم بذلك تلقائيًا مع auth=
r = requests.get(
    "https://api.example.com/me",
    auth=("alice", "mot_de_passe")
)

# الترويسة الخام المكافئة:
# Authorization: Basic YWxpY2U6bW90X2RlX3Bhc3Nl

from requests.auth import HTTPBasicAuth

r = requests.get(url, auth=HTTPBasicAuth("user", "pwd"))

TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

r = requests.get(
    "https://api.github.com/user",
    headers={"Authorization": f"Bearer {TOKEN}"}
)

print(r.json())

import jwt   # pip install PyJWT

token = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjMifQ.signature"

# فك التشفير بدون التحقق (للتصحيح)
payload = jwt.decode(token, options={"verify_signature": False})
print(payload)
# {'sub': '123', 'exp': 1700000000}

# فك التشفير مع التحقق
payload = jwt.decode(token, key="mon_secret", algorithms=["HS256"])

from requests.auth import AuthBase

class BearerAuth(AuthBase):
    def __init__(self, token):
        self.token = token
    def __call__(self, req):
        req.headers["Authorization"] = f"Bearer {self.token}"
        return req

# الاستخدام
r = requests.get(url, auth=BearerAuth(TOKEN))

import time

class TokenManager:
    def __init__(self, client_id, client_secret):
        self.client_id = client_id
        self.client_secret = client_secret
        self._token = None
        self._expires_at = 0
    
    def get_token(self) -> str:
        if not self._token or time.time() > self._expires_at - 60:
            self._refresh()
        return self._token
    
    def _refresh(self):
        r = requests.post("https://auth.example.com/token", data={
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        })
        r.raise_for_status()
        d = r.json()
        self._token = d["access_token"]
        self._expires_at = time.time() + d["expires_in"]

mgr = TokenManager(...)
r = requests.get(url, headers={"Authorization": f"Bearer {mgr.get_token()}"})