في الدروس السابقة، كنا نخزن المهام في قائمة داخل الذاكرة (in-memory list). هذا الحل بسيط وسريع، لكنه يعاني من مشكلة كبيرة: عند إعادة تشغيل الخادم، تفقد جميع البيانات. تخيل أنك تدير موقعاً لتسجيل المستخدمين، ثم يعيد الخادم تشغيل نفسه، فيختفي كل شيء! هنا يأتي دور قاعدة البيانات: هي مكان دائم لتخزين البيانات، حتى بعد إيقاف التطبيق أو إعادة تشغيله.
في هذا الدرس، سنستخدم SQLite، وهي قاعدة بيانات خفيفة ومدمجة مع بايثون، لا تحتاج إلى تثبيت خادم منفصل. سنقوم بربطها مع FastAPI باستخدام مكتبة SQLAlchemy، وهي أشهر مكتبة للتعامل مع قواعد البيانات في بايثون.
database.db)، مما يسهل نقلها أو نسخها احتياطياً.سنقوم بتعديل مشروعنا السابق (الدرس 16) لاستخدام قاعدة بيانات SQLite بدلاً من القائمة المؤقتة. سنحتاج إلى:
sqlalchemy و databases.افتح الطرفية (Terminal) في مجلد مشروعك، وشغّل الأمر التالي:
pip install sqlalchemy databasesملاحظة: databases هي مكتبة مساعدة تسمح لـ FastAPI بالعمل مع SQLAlchemy بشكل غير متزامن (async).
أنشئ ملفاً جديداً باسم 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: كلاس أساسي سنورث منه نماذج جداولنا.الآن سننشئ ملفاً جديداً باسم 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: يحسن سرعة البحث في هذا العمود.الآن سنعدل ملف 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 تلقائياً.الآن سنعدل دوال 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 لتجنب ذلك.create_all عدة مرات، لا تقلق، SQLAlchemy يتجاهل الجداول الموجودة.التمرين: أضف دالة جديدة في API لتحديث مهمة موجودة (PUT /tasks/{task_id}). يجب أن تستقبل بيانات جديدة (title, description, completed) وتحدث المهمة في قاعدة البيانات. إذا لم توجد المهمة، أعد خطأ 404.
تلميح: استخدم db.query(Task).filter(Task.id == task_id).first() للبحث عن المهمة، ثم عدّل خصائص الكائن واستدعِ db.commit().
في هذا الدرس، تعلمنا كيفية ربط FastAPI بقاعدة بيانات SQLite باستخدام SQLAlchemy. أصبحت بياناتنا الآن دائمة، ولن تفقد عند إعادة تشغيل الخادم. في الدرس القادم، سنتعلم كيفية التحقق من البيانات ومعالجة الأخطاء بشكل أكثر احترافية.
جاري تحميل التقييمات...