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

جدول، دستور و تراکنش#

SQLite موتور پایگاه داده‌ای است که در این نمونه داده را در یک فایل نگه می‌دارد و به سرویس جداگانه نیاز ندارد. ماژول sqlite3 در نصب معمول Python موجود است. SQL زبان بیان عملیات روی دادهٔ جدولی است: INSERT برای افزودن ردیف، SELECT برای خواندن، UPDATE برای تغییر و DELETE برای حذف. ترکیب این چهار کار را معمولاً CRUD، یعنی ایجاد، خواندن، به‌روزرسانی و حذف می‌نامند.

هر ردیف جدول یک رکورد و هر ستون یک ویژگی است. کلید اصلی id شناسهٔ رکورد را می‌دهد. قید NOT NULL مقدار تهی را منع و UNIQUE تکرار نام را رد می‌کند. این قیدها در خود پایگاه داده اجرا می‌شوند؛ حتی اگر دو درخواست نزدیک به هم برسند، بررسی یکتایی فقط به جست‌وجوی قبلی برنامه وابسته نخواهد بود.

تراکنش گروهی از تغییرهاست که با commit() ثبت و با rollback() بازگردانده می‌شود. در نمونه‌های این شب از رفتار معمول اتصال sqlite3 استفاده می‌کنیم و ثبت را صریح انجام می‌دهیم. بستن اتصال جای ثبت موفق را نمی‌گیرد. همچنین with connection در sqlite3 مدیریت تراکنش است و لزوماً اتصال را نمی‌بندد؛ به همین علت در کدهای زیر finally داریم.

تمرین مستقل پایگاه داده#

پوشهٔ ~/python-backend/night-09 را بسازید. پیش از اجرای API، فایل inventory_db.py زیر را اجرا کنید. پایگاه آزمایشی آن practice.db است تا با پایگاه API مخلوط نشود. INSERT OR IGNORE در این نمونهٔ کنترل‌شده مانع تکرار رکورد اولیه می‌شود؛ برای ثبت واقعی کاربر بهتر است تعارض را آشکار گزارش کنیم، نه اینکه هر خطای قید را ساکت نادیده بگیریم.

import sqlite3

connection = sqlite3.connect("practice.db")
connection.row_factory = sqlite3.Row
try:
    connection.execute("""
        CREATE TABLE IF NOT EXISTS servers (
            id INTEGER PRIMARY KEY, hostname TEXT NOT NULL UNIQUE
        )
    """)
    connection.execute("INSERT OR IGNORE INTO servers (hostname) VALUES (?)",
                       ("pve01",))
    connection.commit()
    for row in connection.execute("SELECT id, hostname FROM servers ORDER BY id"):
        print(row["id"], row["hostname"])
finally:
    connection.close()

python3 inventory_db.py
python3 inventory_db.py

در هر دو اجرا یک رکورد انتظار می‌رود. علامت ? جای مقدار است و آرگومان دوم مقدار را جدا می‌دهد. کامای انتهای ("pve01",) آن را تاپل تک‌عضوی می‌کند. هرگز متن کاربر را با f-string داخل SQL نگذارید؛ نقل‌قول می‌تواند معنای دستور را تغییر دهد. پارامترها مقدارها را از ساختار SQL جدا نگه می‌دارند. برای نام جدول و ستون از ورودی آزاد استفاده نمی‌کنیم.

برنامهٔ کامل با اتصال مستقل#

در همین پوشه محیط مجازی بسازید و fastapi[standard] را نصب کنید. فایل main.py زیر برنامهٔ مستقل این شب است. DATABASE_PATH از محیط خوانده می‌شود و پیش‌فرض فایل محلی دارد. مقداردهی اولیه را به چرخهٔ آغاز و پایان برنامه یا lifespan منتقل کرده‌ایم؛ بنابراین صرف import کردن ماژول، فایل داده نمی‌سازد و آزمون‌ها فرصت انتخاب پایگاه موقت دارند.

import os
import sqlite3
from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, IPvAnyAddress

DATABASE_PATH = os.getenv("DATABASE_PATH", "inventory.db")


def get_connection():
    connection = sqlite3.connect(DATABASE_PATH, timeout=2)
    connection.row_factory = sqlite3.Row
    return connection


def initialize_database():
    Path(DATABASE_PATH).parent.mkdir(parents=True, exist_ok=True)
    connection = get_connection()
    try:
        connection.execute("""
            CREATE TABLE IF NOT EXISTS servers (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                hostname TEXT NOT NULL UNIQUE,
                address TEXT NOT NULL,
                role TEXT NOT NULL,
                online INTEGER NOT NULL CHECK (online IN (0, 1))
            )
        """)
        connection.commit()
    finally:
        connection.close()


@asynccontextmanager
async def lifespan(app: FastAPI):
    initialize_database()
    yield


app = FastAPI(title="Persistent Inventory API", lifespan=lifespan)


class ServerCreate(BaseModel):
    hostname: str = Field(min_length=1, max_length=63)
    address: IPvAnyAddress
    role: str = Field(min_length=1)
    online: bool = True


def serialize(row):
    record = dict(row)
    record["online"] = bool(record["online"])
    return record


@app.get("/health")
def read_health():
    connection = None
    try:
        connection = get_connection()
        connection.execute("SELECT id FROM servers LIMIT 1").fetchone()
        return {"status": "ok", "database": "ready"}
    except sqlite3.Error:
        raise HTTPException(status_code=503, detail="Database unavailable")
    finally:
        if connection is not None:
            connection.close()


@app.get("/servers")
def read_servers():
    connection = get_connection()
    try:
        rows = connection.execute("SELECT * FROM servers ORDER BY id").fetchall()
        return {"servers": [serialize(row) for row in rows], "total": len(rows)}
    finally:
        connection.close()


@app.get("/servers/{server_id}")
def read_server(server_id: int):
    connection = get_connection()
    try:
        row = connection.execute("SELECT * FROM servers WHERE id = ?",
                                 (server_id,)).fetchone()
        if row is None:
            raise HTTPException(status_code=404, detail="Server not found")
        return serialize(row)
    finally:
        connection.close()


@app.post("/servers", status_code=201)
def create_server(server: ServerCreate):
    connection = get_connection()
    try:
        cursor = connection.execute("""
            INSERT INTO servers (hostname, address, role, online)
            VALUES (?, ?, ?, ?)
        """, (server.hostname, str(server.address), server.role, server.online))
        connection.commit()
        row = connection.execute("SELECT * FROM servers WHERE id = ?",
                                 (cursor.lastrowid,)).fetchone()
        return serialize(row)
    except sqlite3.IntegrityError as error:
        connection.rollback()
        if error.sqlite_errorname == "SQLITE_CONSTRAINT_UNIQUE":
            raise HTTPException(status_code=409, detail="Hostname already exists")
        raise
    finally:
        connection.close()

هر مسیر اتصال خودش را باز و در finally می‌بندد؛ اتصال سراسری مشترک بین درخواست‌ها نداریم. sqlite3.Row خواندن ستون با نام را ممکن می‌کند و dict(row) آن را برای پاسخ آماده می‌سازد. SQLite بولی را به شکل صفر و یک نگه می‌دارد، ولی serialize() قرارداد پاسخ را به true و false برمی‌گرداند. این تصمیم مانع وابسته‌شدن کلاینت به نمایش داخلی پایگاه می‌شود.

در این طرح، IntegrityError یکتایی به 409 تبدیل می‌شود و سایر شکست‌های قید دوباره بالا می‌روند. همهٔ خطاهای پایگاه را «نام تکراری» معرفی نمی‌کنیم. cursor.lastrowid شناسهٔ تولیدشدهٔ INSERT همین اتصال را دارد؛ به‌جای len()+1، تولید شناسه را به پایگاه سپرده‌ایم. این نمونه هنوز راهکار همهٔ بارهای هم‌زمان یا جایگزین طراحی ظرفیت نیست.

بررسی ماندگاری و تمرین#

برنامه را اجرا و درخواست زیر را ارسال کنید. مدل این شب درگاه ندارد، چون رکورد فهرست زیرساخت را نگه می‌داریم؛ مدل شب هشتم یک مثال مستقل ثبت سرویس بود. آدرس و نقش همچنان بخشی از قراردادند. اولین پایگاه تازه باید شناسهٔ یک تولید کند و حالت پیش‌فرض online روشن باشد.

fastapi dev main.py

curl -i -X POST http://127.0.0.1:8000/servers \
  -H 'Content-Type: application/json' \
  -d '{"hostname":"pve01","address":"192.168.10.11","role":"hypervisor"}'
curl http://127.0.0.1:8000/servers
curl -i http://127.0.0.1:8000/servers/999

پس از دریافت 201، سرور را متوقف و از همان پوشه دوباره اجرا کنید؛ GET باید رکورد را نگه داشته باشد. POST تکراری 409 و شناسهٔ ناموجود 404 می‌دهد. برای دیدن مستقیم فایل، در صورت نصب ابزار خط فرمان از sqlite3 inventory.db استفاده کنید؛ دستورهای .tables، .schema servers و SELECT * FROM servers; برای بررسی مفیدند و .quit خارج می‌شود.

تمرین منبع را در checks_db.py و سپس checks_api.py انجام دهید: جدولی با شناسه، نام یکتا، URL، مهلت و وضعیت فعال بسازید. ثبت باید پارامتری، همراه commit و رسیدگی به تعارض باشد. خواندن یک بررسی ناموجود 404 بدهد و داده پس از راه‌اندازی دوباره بماند. URL کامل HTTP/HTTPS ذخیره کنید، نه فقط مسیری مانند /api/health که بدون میزبان مقصد مشخصی ندارد.

برای تمرین SQL، روی یک رکورد آزمایشی با شناسهٔ مشخص، UPDATE پارامتری برای خاموش‌کردن و سپس SELECT برای تأیید بنویسید. اگر حذف را تمرین می‌کنید، فقط همان رکورد ساختگی را با شرط شناسه حذف و نتیجه را بررسی کنید؛ DELETE بدون WHERE همهٔ ردیف‌ها را حذف می‌کند. پیش از این تمرین از فایل آزمایشی نسخهٔ پشتیبان بگیرید.

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

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

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

منابع#