بايثون لأتمتة DevOps

HTTP وواجهات REST API مع مكتبة requests

22 دقيقة الدرس 4 من 28

HTTP وواجهات REST API مع مكتبة requests

البنية التحتية الحديثة تُعرَّف وتُتحكّم بها عبر واجهات API. تُجهّز نسخ EC2 عبر AWS EC2 API، وتفتح طلبات السحب عبر GitHub REST API، وتُنبّه المهندس المناوب عبر PagerDuty API، وتستعلم المقاييس عبر Datadog API. كل سكريبت أتمتة عمليات ستكتبه في يوم ما إما يستدعي API أو يُشغَّل بواسطتها. يجعلك هذا الدرس متقناً لهذا العمل: كيف تستدعي APIs بشكل موثوق، وكيف تُصادق بشكل صحيح، وكيف تنجو من الأعطال العابرة بإعادة المحاولة، وكيف تستهلك الاستجابات المقسّمة على صفحات دون استنزاف الذاكرة.

مكتبة requests والمبادئ الأساسية

مكتبة requests هي المعيار الفعلي لـ HTTP في Python. ثبّتها في venv النشط، ثم أجرِ استدعاء GET بسيطاً:

pip install requests

import requests

# verify=True (التحقق من شهادة TLS) هو الإعداد الافتراضي — لا تعطّله أبداً
response = requests.get("https://api.github.com/repos/torvalds/linux")
response.raise_for_status()   # يرفع HTTPError عند استجابات 4xx / 5xx

data = response.json()        # يُجرجع جسم JSON تلقائياً
print(data["stargazers_count"])

raise_for_status() هي العادة الأهم على الإطلاق. بدونها، يُعيد استجابة 404 أو 500 كائن Response بصمت ويستمر سكريبتك كأن شيئاً لم يحدث — ليفشل لاحقاً بخطأ KeyError محيّر. استدعها دائماً فوراً بعد كل طلب.

مصيدة إنتاجية — verify=False: ستجد أمثلة عبر الإنترنت تمرر verify=False لتخطي التحقق من TLS. في الإنتاج هذه ثغرة أمنية حرجة تُعرّض سكريبتك لهجمات الرجل في المنتصف. إن كنت تتحدث إلى API داخلية بشهادة موقّعة ذاتياً، مرّر verify="/path/to/internal-ca.crt" بدلاً من ذلك. لا تعطّل التحقق من الشهادة أبداً بالكامل.

أنماط المصادقة

أكثر ثلاثة أنماط مصادقة تُصادفها في أعمال العمليات هي رموز Bearer، ومفاتيح API في رؤوس مخصصة، والمصادقة الأساسية. يجب أن تأتي بيانات الاعتماد دائماً من البيئة — ليس من الكود المصدري أبداً.

import os, requests

# --- النمط 1: رمز Bearer (GitHub, Datadog, معظم APIs الحديثة) ---
TOKEN = os.environ["GITHUB_TOKEN"]   # لا تُضمَّن في الكود مطلقاً
headers = {
    "Authorization": f"Bearer {TOKEN}",
    "Accept": "application/vnd.github+json",
    "X-GitHub-Api-Version": "2022-11-28",
}
resp = requests.get("https://api.github.com/user/repos", headers=headers)
resp.raise_for_status()

# --- النمط 2: مفتاح API في رأس مخصص (PagerDuty, Datadog) ---
PD_TOKEN = os.environ["PAGERDUTY_TOKEN"]
pd_headers = {
    "Authorization": f"Token token={PD_TOKEN}",
    "Accept": "application/vnd.pagerduty+json;version=2",
}
resp = requests.get("https://api.pagerduty.com/services", headers=pd_headers)
resp.raise_for_status()

# --- النمط 3: المصادقة الأساسية HTTP (Jenkins, أدوات الاستضافة الذاتية) ---
resp = requests.get(
    "https://jenkins.internal/api/json",
    auth=(os.environ["JENKINS_USER"], os.environ["JENKINS_TOKEN"]),
)
resp.raise_for_status()

استخدم Session للاستدعاءات المتكررة لنفس الخادم: كائن requests.Session() يُعيد استخدام اتصال TCP الأساسي (HTTP keep-alive)، ويشارك ملفات تعريف الارتباط، ويتيح لك ضبط الرؤوس الافتراضية والمصادقة مرة واحدة. على نطاق واسع، إعادة استخدام الاتصال تقلّص وقت الساعة الفعلي لسير عمل مؤلف من 100 طلب بنسبة 30 إلى 60 بالمئة. ضع رؤوس المصادقة على الـ Session لا على كل استدعاء منفرد.

إعادة المحاولة مع التراجع الأسي

الشبكات تفشل. محددات المعدل تُطلق. APIs المنبع تُعيد 503 عابراً. سكريبت يتعطل عند أول فشل ليس جاهزاً للإنتاج. النمط الاحترافي هو إعادة المحاولة مع التراجع الأسي: انتظر 1 ثانية، ثم 2، ثم 4 — مع مكوّن عشوائي صغير (jitter) حتى لا تُعيد جميع السكريبتات المتوازية المحاولة في وقت واحد وتُضاعف الضغط على الخادم.

اربط urllib3.util.Retry بـ HTTPAdapter وثبّته على الـ Session. هذا يعالج إعادة المحاولة على طبقة النقل — بما فيها أعطال مستوى TCP التي لا تُنتج كائن استجابة Python أصلاً.

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

def build_session(token: str) -> requests.Session:
    """أعد Session مُهيّأً مسبقاً بالمصادقة وإعادة المحاولة والمهل الزمنية."""
    retry_strategy = Retry(
        total=5,                            # الحد الأقصى للمحاولات شاملاً الأولى
        backoff_factor=1,                   # أوقات الانتظار: 1 ث، 2 ث، 4 ث، 8 ث، 16 ث
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
        respect_retry_after_header=True,    # احترم Retry-After عند الاستجابة 429
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session = requests.Session()
    session.mount("https://", adapter)
    session.mount("http://",  adapter)
    session.headers.update({
        "Authorization": f"Bearer {token}",
        "Accept": "application/json",
    })
    session.timeout = (5, 30)  # (مهلة الاتصال بالثواني، مهلة القراءة بالثواني)
    return session

session = build_session(os.environ["API_TOKEN"])
resp = session.get("https://api.example.com/resources")
resp.raise_for_status()
print(resp.json())

احترم Retry-After عند الاستجابة 429: حين يُعيد الخادم 429 Too Many Requests، غالباً ما يتضمن رأس Retry-After. ضبط respect_retry_after_header=True (الافتراضي في urllib3 الحديث) يأمر المُهيّئ بالنوم بالضبط بقدر ما يطلب الخادم قبل إعادة المحاولة. هذا هو السلوك الصحيح — لا إعادة اضرب الخادم فوراً وخطر الحظر المؤقت.

التصفّح: متابعة جميع الصفحات دون استنزاف الذاكرة

APIs الإنتاجية لا تُعيد عشرات الآلاف من السجلات في استجابة واحدة. إنها تُقسّم النتائج على صفحات. أكثر أسلوبين شيوعاً هما التصفّح القائم على المؤشر (حديث، توصي به GitHub وStripe) والتصفّح القائم على رقم الصفحة (أقدم، لا يزال في كل مكان). كلاهما يستلزم التكرار حتى تُشير API بلا مزيد من الصفحات.

دورة حياة استدعاء API الكاملة: Session مع HTTPAdapter يعالج إعادة المحاولة والتراجع بشفافية بينما تتكرر طبقة التطبيق عبر النتائج المقسّمة بالمؤشر.

import os, requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from typing import Generator

def build_session(token: str) -> requests.Session:
    retry = Retry(total=5, backoff_factor=1, status_forcelist=[429,500,502,503,504])
    s = requests.Session()
    s.mount("https://", HTTPAdapter(max_retries=retry))
    s.headers.update({"Authorization": f"Bearer {token}", "Accept": "application/json"})
    s.timeout = (5, 30)
    return s

# --- التصفّح القائم على المؤشر (GitHub, Stripe, Slack) ---
def iter_github_repos(session: requests.Session, org: str) -> Generator[dict, None, None]:
    """أعِد كل مستودع في منظمة بمتابعة رؤوس Link بشفافية."""
    url = f"https://api.github.com/orgs/{org}/repos"
    params: dict = {"per_page": 100}   # اطلب دائماً أقصى حجم صفحة
    while url:
        resp = session.get(url, params=params)
        resp.raise_for_status()
        yield from resp.json()
        # requests تُجرجع رأس Link تلقائياً في resp.links
        url = resp.links.get("next", {}).get("url")
        params = {}   # URL التالي يحتوي بالفعل على المعاملات

# --- التصفّح القائم على رقم الصفحة (Jenkins, APIs أقدم) ---
def iter_all_pages(session: requests.Session, base_url: str) -> Generator[dict, None, None]:
    page = 1
    while True:
        resp = session.get(base_url, params={"page": page, "per_page": 100})
        resp.raise_for_status()
        items = resp.json()
        if not items:   # قائمة فارغة تدل على الصفحة الأخيرة
            break
        yield from items
        page += 1

session = build_session(os.environ["GITHUB_TOKEN"])
for repo in iter_github_repos(session, "myorg"):
    print(repo["full_name"], repo["stargazers_count"])

المولّدات تُبقي الذاكرة ثابتة: كلتا دالتَي التصفّح تستخدمان yield from بدلاً من بناء قائمة. المولّد يُنتج صفحة واحدة في كل مرة فيبقى استخدام الذاكرة محدوداً بحجم الصفحة الواحدة بصرف النظر عن العدد الإجمالي. منظمة بعشرة آلاف مستودع في قائمة قد تستهلك مئات الميغابايت؛ النسخة بالمولّد تبقى ثابتة.

إرسال البيانات: POST وPUT وPATCH

العمليات الكتابية — إنشاء حادثة، تشغيل نشر، نشر رسالة Slack — تستخدم POST أو PUT مع جسم JSON. مرّر قاموس Python لمعامل json= وتُجرجع requests وتضبط Content-Type: application/json تلقائياً:

# تشغيل حادثة PagerDuty عبر POST
payload = {
    "incident": {
        "type": "incident",
        "title": "استخدام القرص فوق 90% على prod-db-01",
        "service": {"id": os.environ["PD_SERVICE_ID"], "type": "service_reference"},
        "urgency": "high",
        "body": {
            "type": "incident_body",
            "details": "تنبيه تلقائي من سكريبت مراقبة القرص.",
        },
    }
}
resp = session.post("https://api.pagerduty.com/incidents", json=payload)
resp.raise_for_status()
incident_id = resp.json()["incident"]["id"]
print(f"تم إنشاء الحادثة {incident_id}")

مفاتيح القابلية للتكرار في العمليات الكتابية: بعض APIs تقبل رأس Idempotency-Key (Stripe, PagerDuty). مرّر قيمة حتمية — كهاش محتوى التنبيه زائد ساعة UTC الحالية مقرّبة — حتى إن أعاد سكريبتك المحاولة بعد انتهاء مهلة الشبكة، يُكرّر الخادم الطلب ولا يُنشئ حادثتين. هذا هو الحل الصحيح لمشكلة "إعادة المحاولة أنشأت نسخة مكررة" — لا تعطيل إعادة المحاولة.

المهل الزمنية ومعالجة الأخطاء المنظَّمة

أكثر سبب شائع لتجمّد سكريبتات العمليات إلى الأبد في الإنتاج هو مهلة زمنية مفقودة. اضبط مهلة الاتصال ومهلة القراءة صراحةً. إعداد جيد افتراضي هو (5, 30): خمس ثوانٍ لإنشاء اتصال TCP، ثلاثون ثانية لاستقبال كامل جسم الاستجابة. التقط أنواع الاستثناءات المحددة لتُخبر المشغّل بدقة ما الذي أخطأ:

import requests.exceptions as exc

try:
    resp = session.get("https://api.example.com/endpoint")
    resp.raise_for_status()
except exc.ConnectTimeout:
    print("تعذّر الوصول للخادم خلال 5 ث — تحقق من الشبكة / جدار الحماية")
    raise SystemExit(1)
except exc.ReadTimeout:
    print("اتصل الخادم لكنه لم يستجب خلال 30 ث")
    raise SystemExit(1)
except exc.HTTPError as e:
    print(f"HTTP {e.response.status_code}: {e.response.text[:200]}")
    raise SystemExit(1)
except exc.RequestException as e:
    # يلتقط ConnectionError وTooManyRedirects وكل ما ترفعه requests
    print(f"خطأ شبكة: {e}")
    raise SystemExit(1)