طلبات HTTP باستخدام requests وhttpx

مكتبة requests

# pip install requests
import requests

r = requests.get("https://api.github.com/users/octocat")
print(r.status_code)   # 200
print(r.json()["login"])   # octocat

معلمات الاستعلام والرؤوس

import requests

r = requests.get(
    "https://api.example.com/search",
    params={"q": "python", "page": 1},
    headers={"Authorization": "Bearer my-token"}
)
print(r.url)    # full URL with query string

POST مع جسم JSON

import requests

r = requests.post(
    "https://api.example.com/users",
    json={"name": "Alice", "email": "alice@example.com"}
)
print(r.status_code)   # 201
print(r.json())

كائنات الجلسات

import requests

with requests.Session() as s:
    s.headers.update({"Authorization": "Bearer token"})
    users = s.get("https://api.example.com/users").json()
    profile = s.get("https://api.example.com/me").json()

المهلة وإعادة المحاولة

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

s = requests.Session()
retry = Retry(total=3, backoff_factor=1)
s.mount("https://", HTTPAdapter(max_retries=retry))

r = s.get("https://api.example.com/data", timeout=5)

معالجة الأخطاء

import requests

try:
    r = requests.get("https://api.example.com/data")
    r.raise_for_status()
    data = r.json()
except requests.HTTPError as e:
    print(f"HTTP error: {e.response.status_code}")
except requests.ConnectionError:
    print("Network unreachable")

تنزيل الملفات

import requests

with requests.get("https://example.com/large.zip", stream=True) as r:
    r.raise_for_status()
    with open("large.zip", "wb") as f:
        for chunk in r.iter_content(chunk_size=8192):
            f.write(chunk)

‏httpx — HTTP غير متزامن حديث

# pip install httpx
import httpx

# Synchronous:
r = httpx.get("https://httpbin.org/get")
print(r.json())

# Asynchronous:
import asyncio
async def main():
    async with httpx.AsyncClient() as c:
        r = await c.get("https://httpbin.org/get")
        print(r.json())
asyncio.run(main())

‏httpx AsyncClient

import httpx, asyncio

CLIENT: httpx.AsyncClient | None = None

async def startup():
    global CLIENT
    CLIENT = httpx.AsyncClient(timeout=10)

async def shutdown():
    await CLIENT.aclose()

مساعدات المصادقة

import httpx

class BearerAuth(httpx.Auth):
    def __init__(self, token): self.token = token
    def auth_flow(self, request):
        request.headers["Authorization"] = f"Bearer {self.token}"
        yield request

async with httpx.AsyncClient(auth=BearerAuth("my-token")) as c:
    r = await c.get("https://api.example.com/me")

محاكاة HTTP في الاختبارات

# pip install respx
import httpx, respx, asyncio

@respx.mock
async def test_api():
    respx.get("https://api.example.com/users").mock(
        return_value=httpx.Response(200, json=[{"id":1}])
    )
    async with httpx.AsyncClient() as c:
        r = await c.get("https://api.example.com/users")
    assert r.json() == [{"id": 1}]

تحقق سريع

مراجعة