طلبات HTTP وواجهات API

15.1 الطلب الأول – הבקשה הראשונה (The First Request)

اللوحة الآن عميل. والمكتبة requests موجودة مسبقا في نسخ مايكروبايثون ذات الشبكة على اللوحتين، وهي تفتح المقبس وتكتب الطلب وتقرأ الاستجابة عنا.

import requests
import wifi

wifi.connect()
r = requests.get("http://api.example.com/v1/hello")
print(r.status_code)
print(r.text)
r.close()

200
hello from the server

العدد الأول هو رمز الحالة (קוד סטטוס). هو أول ما نقرؤه، لأن الخادم يجيب حتى حين يرفض طلبنا:

15.2 خدمة الطقس: قراءة JSON – שירות מזג אוויר: קריאת JSON (A Weather Service: Reading JSON)

تعرّفنا إلى JSON في فصل الملفات. والأمر r.json() يحوّل الجواب إلى قاموس جاهز. نضع هنا timeout، ونغلق الاستجابة في finally، وهذا الشكل سنكرره دائما. عنوان الخدمة في المثال ليس خدمة فعلية، والجواب من العالم المُحاكى.

import requests
import wifi

URL = "http://api.example.com/v1/weather?city=Haifa"

wifi.connect()
r = requests.get(URL, timeout=5)
try:
    if r.status_code == 200:
        data = r.json()
        print(data["city"])
        print("temp: %.1f C" % data["temp_c"])
        print("wind: %.1f km/h" % (data["wind_ms"] * 3.6))
    else:
        print("server said", r.status_code)
finally:
    r.close()

Haifa
temp: 24.5 C
wind: 18.0 km/h

15.3 إرسال البيانات والمفاتيح – שליחת נתונים ומפתחות (Sending Data, API Keys)

الأمر post يرسل بيانات في جسم الطلب. والمعامل json= يحوّل القاموس إلى نص ويضع الترويسة المناسبة.

import board
import requests
import wifi

URL = "http://api.example.com/v1/readings"

wifi.connect()
pot = board.analog_in(board.POT)
reading = {"sensor": "pot", "raw": pot.read_u16()}
r = requests.post(URL, json=reading, timeout=5)
try:
    print(r.status_code, r.text)
finally:
    r.close()

201 {"id":17}
[http_sent]
POST http://api.example.com/v1/readings {'raw': 32776, 'sensor': 'pot'}

أغلب الخدمات الحقيقية تطلب مفتاحا (מפתח API) يعرّف صاحب الحساب. نضيفه إلى secrets.py سطرا جديدا، ونقرؤه كما نقرأ كلمة مرور الشبكة. المقطع التالي مرّ بالترجمة فقط ولم يُشغَّل في بناء الكتاب، لأن ملف الأسرار في الكتاب لا يحوي مفتاحا:

import requests
from secrets import API_KEY

headers = {"Authorization": "Bearer " + API_KEY}
r = requests.get(URL, headers=headers, timeout=5)

15.4 حين لا يسير الطلب كما نتمنى – טיפול בשגיאות ובפסק זמן (Errors and Timeouts)

على المتحكّم الذي يعمل وحده لا أحد يعيد تشغيل البرنامج. لذلك نميّز أربع حالات:

نجمع المعالجة في دالة واحدة تعيد البيانات أو None، فلا يحتاج باقي البرنامج إلى التفكير في ذلك:

import requests
import wifi

BASE = "http://api.example.com/v1/"


def get_json(url):
    try:
        r = requests.get(url, timeout=5)
    except OSError as e:
        print("network error:", e)
        return None
    try:
        if r.status_code != 200:
            print("status:", r.status_code)
            return None
        return r.json()
    except ValueError:
        print("bad JSON")
        return None
    finally:
        r.close()


wifi.connect()
for name in ("hello", "busy", "broken", "gone"):
    print(name, "->", get_json(BASE + name))

hello -> {'ok': True}
status: 500
busy -> None
bad JSON
broken -> None
network error: -202
gone -> None

15.5 الساعة من الشبكة: NTP – שעון מהרשת: NTP (Network Time)

لا تعرف اللوحة كم الساعة حين تبدأ، إذ لا بطارية عادة لساعتها الداخلية. والبروتوكول NTP يسأل خادم وقت في الإنترنت ويضبطها. الأمر ntptime.settime() يضع في الساعة التوقيت العالمي UTC، وعلينا نحن إضافة فرق المنطقة: في إسرائيل ساعتان في الشتاء، وثلاث ساعات في الصيف.

import ntptime
import time
import wifi

OFFSET_S = 3 * 3600

wifi.connect()
try:
    ntptime.settime()
except OSError as e:
    print("no time:", e)

t = time.gmtime()
print("UTC:   %04d-%02d-%02d %02d:%02d" % t[:5])
t = time.gmtime(time.time() + OFFSET_S)
print("local: %04d-%02d-%02d %02d:%02d" % t[:5])

UTC:   2027-01-01 00:00
local: 2027-01-01 03:00

15.6 الجواب الكبير والذاكرة – תשובה גדולה וזיכרון (Large Responses and Memory)

الجواب كله يدخل إلى الذاكرة: r.content يقرأ الجواب كاملا مرة واحدة، وr.json() يبني منه قاموسا يأخذ مكانا إضافيا. وعلى المتحكّم تنتهي الذاكرة بخطأ MemoryError. كمية الذاكرة الحرة نقيسها على لوحتنا بـgc.mem_free()، فهي تتغيّر بحسب البرنامج.

import gc
import requests
import wifi

URL = "http://api.example.com/v1/forecast?days=7"

wifi.connect()
gc.collect()
r = requests.get(URL, timeout=5)
try:
    print("bytes:", len(r.content))
    hours = r.json()["hours"]
    print("items:", len(hours))
    print("first: %.1f C" % hours[0]["temp_c"])
finally:
    r.close()
del hours
gc.collect()

bytes: 6285
items: 168
first: 20.5 C

أفضل علاج هو أن نطلب أقل: كثير من الخدمات تقبل معاملا يحدّد الأيام أو الحقول. وإن لم يكن، نقرأ الجواب قطعا بقطع من r.raw دون تخزينه كله. وهذا المقطع مرّ بالترجمة فقط ولم يُشغَّل في بناء الكتاب، لأن العالم المُحاكى لا يوفّر r.raw:

r = requests.get(URL, timeout=5)
total = 0
while True:
    chunk = r.raw.read(256)
    if not chunk:
        break
    total += len(chunk)
r.close()
print(total)

15.7 HTTPS وخدمات أخرى – HTTPS ושירותים אחרים

حسب العنوان يختار requests الطريق: يكفي أن نكتب https:// ليصبح الطلب مشفّرا، وهذا يحمي المحتوى، بما فيه المفتاح، ممن يتنصّت على الشبكة. لكن له ثمن: مصافحة التشفير تأخذ زمنا وذاكرة أكبر بكثير من طلب عادي.

وكثيرا ما يُطلب إرسال القراءات إلى جدول، مثل Google Sheets، أو إلى منصة بيانات. الفكرة في كل هذه الخدمات واحدة: تعطينا الخدمة عنوانا ومفتاحا، وترسل اللوحة post بصيغة JSON عبر HTTPS كل فترة، بالشكل الذي عرفناه في هذا الفصل. والتفاصيل تتغيّر من خدمة إلى أخرى، فنتبع توثيقها، ولا نعرضها هنا.

وعلى Raspberry Pi 4/5 البرنامج أقصر: نظام التشغيل هو الذي يتصل بالشبكة، ولذا لا نحتاج إلى wifi.py، والمكتبة requests هي الأصل الذي تقلّده مكتبة مايكروبايثون. نثبّتها في بيئة خاصة. البرنامج التالي مرّ بالترجمة فقط ولم يُشغَّل في بناء الكتاب، لأن مُحاكاة الباي لا تحوي شبكة:

python3 -m venv ~/lab
~/lab/bin/pip install requests
~/lab/bin/python weather_pi.py

import requests

URL = "http://api.example.com/v1/weather?city=Haifa"

with requests.get(URL, timeout=5) as r:
    r.raise_for_status()
    data = r.json()
print("temp: %.1f C" % data["temp_c"])

أهمّ أفكار الفصل:

• نقرأ رمز الحالة أولا: الخادم يجيب حتى حين يرفض.
• كل استجابة تُغلق في finally، وكل طلب له timeout.
• أربع حالات فشل مختلفة، ونجمع معالجتها في دالة واحدة.
• المفاتيح في secrets.py وحده، والمفتاح المتسرّب يُلغى.
• NTP يعطي UTC، وفرق المنطقة علينا.
• نقدّر حجم الجواب قبل طلبه، ونطلب أقل ما نحتاج إليه.
• HTTPS يخفي المحتوى، وهو أثقل، ولا يضمن وحده هوية الخادم.

أوامر الفصل:

قاموس المصطلحات – מילון מונחים:

أسئلة للتمرين:

from machine import Pin
import board
import requests
import wifi

URL = "http://api.example.com/v1/weather?city=Haifa"
LIMIT_KMH = 30

led = Pin(board.LED, Pin.OUT)
wifi.connect()
r = requests.get(URL, timeout=5)
try:
    kmh = r.json()["wind_ms"] * 3.6
finally:
    r.close()
led.value(1 if kmh > LIMIT_KMH else 0)
print("%.1f km/h, LED %d" % (kmh, led.value()))

import time


def get_retry(url, tries=3, wait_s=2):
    for k in range(tries):
        data = get_json(url)
        if data is not None:
            return data
        if k < tries - 1:
            time.sleep(wait_s)
    return None