ربط FastAPI بقاعدة بيانات SQLite

الدرس 17: ربط FastAPI بقاعدة بيانات SQLite

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

في الدروس السابقة، كنا نخزن المهام في قائمة داخل الذاكرة (in-memory list). هذا الحل بسيط وسريع، لكنه يعاني من مشكلة كبيرة: عند إعادة تشغيل الخادم، تفقد جميع البيانات. تخيل أنك تدير موقعاً لتسجيل المستخدمين، ثم يعيد الخادم تشغيل نفسه، فيختفي كل شيء! هنا يأتي دور قاعدة البيانات: هي مكان دائم لتخزين البيانات، حتى بعد إيقاف التطبيق أو إعادة تشغيله.

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

لماذا SQLite؟

• سهلة التثبيت: لا تحتاج إلى تثبيت برنامج إضافي، فهي جزء من بايثون.
• مناسبة للمشاريع الصغيرة: مثالية للتطبيقات التي تعمل على جهاز واحد أو لاختبار الأفكار.
• ملف واحد: قاعدة البيانات كلها موجودة في ملف واحد (مثلاً database.db)، مما يسهل نقلها أو نسخها احتياطياً.

خطوات التنفيذ

سنقوم بتعديل مشروعنا السابق (الدرس 16) لاستخدام قاعدة بيانات SQLite بدلاً من القائمة المؤقتة. سنحتاج إلى:

1. تثبيت المكتبات اللازمة: sqlalchemy و databases.
2. إنشاء ملف لتوصيل قاعدة البيانات (database.py).
3. إنشاء نموذج قاعدة البيانات (Model) باستخدام SQLAlchemy.
4. تعديل دوال CRUD لقراءة وكتابة البيانات من قاعدة البيانات.

تثبيت المكتبات

افتح الطرفية (Terminal) في مجلد مشروعك، وشغّل الأمر التالي:

pip install sqlalchemy databases

ملاحظة: databases هي مكتبة مساعدة تسمح لـ FastAPI بالعمل مع SQLAlchemy بشكل غير متزامن (async).

إنشاء ملف توصيل قاعدة البيانات (database.py)

أنشئ ملفاً جديداً باسم database.py في مجلد app، وأضف الكود التالي:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# عنوان قاعدة البيانات SQLite
SQLALCHEMY_DATABASE_URL = "sqlite:///./tasks.db"

# إنشاء محرك قاعدة البيانات
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

# إنشاء جلسة للتعامل مع قاعدة البيانات
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# قاعدة النماذج (Base class) لتعريف جداول قاعدة البيانات
Base = declarative_base()

شرح الكود:

• create_engine: ينشئ اتصالاً بقاعدة البيانات. المعامل connect_args={"check_same_thread": False} ضروري لـ SQLite لأنه يسمح بمشاركة الاتصال بين عدة خيوط (threads).
• SessionLocal: دالة تنشئ جلسة جديدة لكل طلب (request). سنستخدمها لاحقاً في دوال API.
• Base: كلاس أساسي سنورث منه نماذج جداولنا.

إنشاء نموذج قاعدة البيانات (Model)

الآن سننشئ ملفاً جديداً باسم models.py داخل مجلد app، ونضيف فيه تعريف جدول المهام:

from sqlalchemy import Column, Integer, String, Boolean
from database import Base

class Task(Base):
    __tablename__ = "tasks"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, default="")
    completed = Column(Boolean, default=False)

شرح الكود:

• __tablename__: اسم الجدول في قاعدة البيانات.
• Column: تمثل عموداً في الجدول. نحدد نوع البيانات (Integer, String, Boolean).
• primary_key=True: يجعل العمود مفتاحاً أساسياً (فريداً لكل سجل).
• index=True: يحسن سرعة البحث في هذا العمود.

تعديل دوال CRUD لاستخدام قاعدة البيانات

الآن سنعدل ملف main.py (أو crud.py إذا كنت قد نظمت المشروع) لاستخدام قاعدة البيانات بدلاً من القائمة. أولاً، نضيف دوال مساعدة للحصول على جلسة قاعدة البيانات:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import SessionLocal, engine
from models import Base, Task
from pydantic import BaseModel

# إنشاء جداول قاعدة البيانات (إذا لم تكن موجودة)
Base.metadata.create_all(bind=engine)

app = FastAPI()

# دالة للحصول على جلسة قاعدة البيانات لكل طلب
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# نموذج Pydantic للإدخال والإخراج
class TaskCreate(BaseModel):
    title: str
    description: str = ""
    completed: bool = False

class TaskResponse(TaskCreate):
    id: int

    class Config:
        orm_mode = True

شرح الكود:

• Base.metadata.create_all(bind=engine): ينشئ الجداول في قاعدة البيانات عند تشغيل التطبيق لأول مرة.
• get_db: دالة تعتمد عليها FastAPI (باستخدام Depends) لتوفير جلسة قاعدة بيانات لكل طلب، ثم تغلقها تلقائياً بعد انتهاء الطلب.
• orm_mode = True: يسمح لـ Pydantic بتحويل كائنات SQLAlchemy إلى JSON تلقائياً.

تعديل دوال API (مثال: إنشاء مهمة جديدة)

الآن سنعدل دوال CRUD لاستخدام قاعدة البيانات. إليك مثال لدالة إنشاء مهمة جديدة:

@app.post("/tasks/", response_model=TaskResponse)
def create_task(task: TaskCreate, db: Session = Depends(get_db)):
    db_task = Task(**task.dict())
    db.add(db_task)
    db.commit()
    db.refresh(db_task)
    return db_task

شرح الكود:

• db: Session = Depends(get_db): يحقن جلسة قاعدة البيانات في الدالة.
• Task(**task.dict()): ينشئ كائن SQLAlchemy جديد من بيانات Pydantic.
• db.add(db_task): يضيف الكائن إلى الجلسة.
• db.commit(): يحفظ التغييرات في قاعدة البيانات.
• db.refresh(db_task): يحدث الكائن بأي قيم افتراضية من قاعدة البيانات (مثل id).

أمثلة لدوال أخرى (قراءة، تحديث، حذف)

إليك دوال قراءة جميع المهام وقراءة مهمة واحدة:

@app.get("/tasks/", response_model=list[TaskResponse])
def read_tasks(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    tasks = db.query(Task).offset(skip).limit(limit).all()
    return tasks

@app.get("/tasks/{task_id}", response_model=TaskResponse)
def read_task(task_id: int, db: Session = Depends(get_db)):
    task = db.query(Task).filter(Task.id == task_id).first()
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

شرح الكود:

• db.query(Task): يبدأ استعلامًا على جدول المهام.
• .offset(skip).limit(limit): يستخدم للتقسيم (pagination).
• .filter(Task.id == task_id): يصفّي النتائج حسب الشرط.
• .first(): يعيد أول نتيجة أو None إذا لم يوجد.

الأخطاء الشائعة

• نسيان استدعاء db.commit(): إذا لم تستدعِ commit، لن تُحفظ التغييرات في قاعدة البيانات.
• عدم إغلاق الجلسة: قد يؤدي إلى تسرب الذاكرة. استخدم دالة get_db مع Depends لتجنب ذلك.
• خطأ "Table already exists": إذا قمت بتشغيل create_all عدة مرات، لا تقلق، SQLAlchemy يتجاهل الجداول الموجودة.

تمرين صغير

الخلاصة

في هذا الدرس، تعلمنا كيفية ربط FastAPI بقاعدة بيانات SQLite باستخدام SQLAlchemy. أصبحت بياناتنا الآن دائمة، ولن تفقد عند إعادة تشغيل الخادم. في الدرس القادم، سنتعلم كيفية التحقق من البيانات ومعالجة الأخطاء بشكل أكثر احترافية.