API پایدار هنوز اجازه می‌دهد هر کلاینت محلی داده ثبت کند. این شب عملیات نوشتن را با کلید دسترسی محدود می‌کنیم، قراردادها را با pytest می‌سنجیم و رویدادهای مهم را برای بهره‌بردار ثبت می‌کنیم. خروجی مورد انتظار برنامه‌ای است که خواندن عمومی و نوشتن کنترل‌شده دارد و می‌توان رفتار آن را بدون راه‌انداختن سرور شبکه آزمود. پیش‌نیاز، برنامهٔ SQLite شب نهم است.

کنترل دسترسی پیش از اجرای مسیر#

dependency در FastAPI تابعی است که چارچوب برای فراهم‌کردن نیاز یک مسیر اجرا می‌کند. اینجا نیاز مسیر نوشتن، تأیید کلید است. APIKeyHeader مقدار هدر را استخراج می‌کند؛ خودش نمی‌داند کلید صحیح چیست. بررسی واقعی را در تابع require_key انجام می‌دهیم. با قرارگرفتن این dependency روی POST، درخواست ردشده به بدنهٔ ثبت و INSERT نمی‌رسد.

پوشهٔ ~/python-backend/night-10 بسازید و فقط main.py شب قبل را به آن کپی کنید؛ پایگاه قبلی را کپی نکنید. محیط مجازی تازه بسازید و fastapi[standard] و pytest را نصب کنید. در فایل جدید security.py کد زیر را قرار دهید. کلید از محیط خوانده می‌شود و اگر وجود نداشته باشد برنامه هنگام import این ماژول متوقف می‌شود؛ بدون تنظیم اعتبار، مسیر نوشتن باز نخواهد ماند.

import os
import secrets
from typing import Annotated

from fastapi import Depends, HTTPException
from fastapi.security import APIKeyHeader

expected_key = os.getenv("INVENTORY_API_KEY")
if not expected_key or not expected_key.isascii():
    raise RuntimeError("Set a non-empty ASCII INVENTORY_API_KEY")

key_header = APIKeyHeader(name="X-API-Key", auto_error=False)


def require_key(key: Annotated[str | None, Depends(key_header)]) -> None:
    if key is None or not secrets.compare_digest(key.encode(), expected_key.encode()):
        raise HTTPException(
            status_code=401,
            detail="Invalid or missing API key",
            headers={"WWW-Authenticate": "APIKey"},
        )

کلید این آزمایشگاه باید رشتهٔ ASCII غیرخالی باشد؛ مقدار تصادفی مناسب انتخاب کنید. مقایسهٔ بایتی با compare_digest() اثر وابستگی زمان مقایسه به محتوای راز را کاهش می‌دهد. این کلید یک اعتبار مشترک ساده است و جای هویت جداگانهٔ کاربران، نقش‌ها و چرخهٔ کامل مدیریت اعتبار را نمی‌گیرد. انتقال آن بیرون از میزبان محلی باید با HTTPS انجام شود.

در importهای main.py، Depends و تابع کنترل را اضافه و decorator موجود POST را جایگزین کنید. تابع create_server و مسیرهای GET همان پیاده‌سازی شب قبل می‌مانند. کلید را به query یا بدنه منتقل نکنید؛ URL ممکن است در تاریخچه و گزارش‌های دسترسی ثبت شود و بدنه نیز قرارداد دادهٔ سرور است.

from fastapi import Depends, FastAPI, HTTPException
from security import require_key

@app.post("/servers", status_code=201, dependencies=[Depends(require_key)])

در ترمینال نخست کلید را بدون نوشتن مقدارش در تاریخچه بگیرید و صادر کنید. export لازم است تا فرایند Python متغیر را دریافت کند. ترمینال دوم این متغیر را خودکار ندارد؛ همان کلید را دوباره در آن وارد کنید. پنهان‌کردن ورودی در ترمینال جای حفاظت از محیط فرایند و میزبان را نمی‌گیرد، ولی از ثبت مستقیم راز در فرمان جلوگیری می‌کند.

read -rsp 'Choose a local API key: ' INVENTORY_API_KEY
echo
export INVENTORY_API_KEY
fastapi dev main.py

با curl، سلامت و خواندن فهرست باید بدون هدر 200 بدهند؛ POST بدون هدر یا با کلید اشتباه 401 است. در ترمینال دوم، کلید صحیح را در API_KEY بگیرید و درخواست ثبت را با -H "X-API-Key: $API_KEY" بفرستید. همان بدنهٔ شب قبل را استفاده کنید. انتظار 201 و باقی‌ماندن رکورد پس از راه‌اندازی دوباره است؛ کلید را در خروجی عیب‌یابی چاپ نکنید.

آزمون باید مستقل و قابل تکرار باشد#

pytest تابع‌هایی با نام test_* را پیدا می‌کند. دستور assert انتظار مشخصی مانند برابری کد پاسخ را می‌سنجد و هنگام شکست، محل و تفاوت را نشان می‌دهد. TestClient درخواست را داخل فرایند به برنامه می‌فرستد؛ نیازی به fastapi dev و درگاه آزاد ندارد. در این نمونه با context manager آن را باز می‌کنیم تا lifespan و ساخت جدول هم اجرا شوند.

فایل test_inventory.py را با نمونهٔ زیر بسازید. fixture با نام client برای هر آزمون پایگاه جدا می‌سازد. tmp_path مسیر موقت اختصاصی آزمون است و monkeypatch مقدار متغیر برنامه را موقت عوض می‌کند. چون شب نهم ساخت پایگاه را هنگام import انجام نمی‌دهد، این تغییر پیش از نخستین اتصال مؤثر خواهد بود.

import pytest
from fastapi.testclient import TestClient

import main

TEST_KEY = "test-only-key"
PAYLOAD = {"hostname": "pve01", "address": "192.168.10.11", "role": "hypervisor"}


@pytest.fixture
def client(tmp_path, monkeypatch):
    monkeypatch.setattr(main, "DATABASE_PATH", str(tmp_path / "test.db"))
    with TestClient(main.app) as test_client:
        yield test_client


def test_public_read(client):
    response = client.get("/servers")
    assert response.status_code == 200
    assert response.json()["total"] == 0


def test_rejected_write_inserts_nothing(client):
    response = client.post("/servers", json=PAYLOAD)
    assert response.status_code == 401
    assert client.get("/servers").json()["total"] == 0


def test_authenticated_write(client):
    response = client.post("/servers", json=PAYLOAD,
                           headers={"X-API-Key": TEST_KEY})
    assert response.status_code == 201
    assert client.get("/servers").json()["servers"][0]["hostname"] == "pve01"


def test_missing_server(client):
    assert client.get("/servers/999").status_code == 404

INVENTORY_API_KEY='test-only-key' python -m pytest -q

چهار آزمون موفق انتظار می‌رود. کلید این فرمان فقط اعتبار ساختگی آزمون است. هیچ آزمونی را به پایگاه واقعی اشاره ندهید؛ اجرای تکراری نباید به دادهٔ قبلی یا ترتیب آزمون‌ها وابسته باشد. برای دیدن گزارش شکست، موقتاً انتظار 404 را به 200 تغییر دهید، آزمون را اجرا و سپس اصلاح را برگردانید. این کار نشان می‌دهد آزمون واقعاً تفاوت قرارداد را می‌بیند.

این آزمون‌ها ماندگاری پس از توقف فرایند را مستقیماً نمی‌سنجند. تمرین تکمیلی این است که دو context جدا از TestClient را با همان فایل موقت باز کنید؛ در اول ثبت و در دوم خواندن انجام دهید. برعکس، برای استقلال آزمون‌های معمول هرکدام باید فایل تازه بگیرند. هدف دو آزمایش متفاوت است و نباید تصادفاً دادهٔ مشترک ایجاد شود.

رویداد برای بهره‌بردار، پاسخ برای کلاینت#

ماژول logging شدت رویداد را مشخص می‌کند و اجازه می‌دهد میزان جزئیات با تنظیم عوض شود. در main.py پیش از تعریف مسیرها کد زیر را اضافه کنید. سطح پیش‌فرض INFO است؛ مقدار نامعتبر را صریح رد می‌کنیم. این تنظیم برای اجرای تک‌برنامه‌ای آزمایشگاه است و گزارش‌های خود Uvicorn ممکن است تنظیم جداگانه داشته باشند.

import logging
import sys

log_level = os.getenv("LOG_LEVEL", "INFO").upper()
if log_level not in {"DEBUG", "INFO", "WARNING", "ERROR"}:
    raise ValueError("Invalid LOG_LEVEL")
logging.basicConfig(level=log_level,
                    format="%(levelname)s %(name)s %(message)s",
                    stream=sys.stdout)
logger = logging.getLogger("inventory")

در تابع ثبت، بعد از commit() موفق، شناسه و نام میزبان را با پیام زیر ثبت کنید. در شاخهٔ تعارض یکتایی، پیش از raise HTTPException از logger.warning("Duplicate hostname=%s", server.hostname) استفاده کنید. مقدارها جدا از قالب پیام ارسال می‌شوند؛ کل بدنه، هدر کلید و رشتهٔ اتصال خصوصی را در پیام نگذارید.

logger.info("Server created hostname=%s id=%s", server.hostname, cursor.lastrowid)

با LOG_LEVEL=INFO ثبت موفق باید پیام INFO و تکرار باید WARNING تولید کند. با LOG_LEVEL=WARNING، ثبت موفق همچنان 201 می‌دهد ولی پیام INFO برنامه نمایش داده نمی‌شود. سطح گزارش، منطق کسب‌وکار یا خروجی تابع را عوض نمی‌کند. ثبت «ساخته شد» باید پس از commit باشد تا گزارش، رویدادی را که هنوز پایدار نشده قطعی معرفی نکند.

تمرین پایانی، افزودن آزمون برای کلید اشتباه، نام تکراری و آدرس نامعتبر است. انتظارها به‌ترتیب 401، 409 و 422 هستند و در هر حالت شمار رکوردها باید مطابق قرارداد بماند. سپس با گزارش‌ها بررسی کنید خطای تکرار دیده می‌شود ولی کلید در هیچ پیام برنامه نیست. این مجموعه، حفاظت و قابلیت بررسی را به‌صورت رفتار قابل مشاهده به هم وصل می‌کند.

در تمرین گزارش، میان خطای انتظاررفته و خطای برنامه تمایز بگذارید. نام تکراری در این قرارداد رویدادی قابل توضیح است و پاسخ مشخص دارد؛ خرابی فایل پایگاه مسئلهٔ دیگری است و نباید با همان پیام پوشیده شود. اگر بعداً پیام خطای داخلی را ثبت کردید، جزئیات حساس را به پاسخ عمومی نفرستید. رویداد باید به تشخیص علت کمک کند و هم‌زمان رازهای اعتبارسنجی را وارد فایل‌های قابل اشتراک نکند.

منابع#