الدرس 20: بناء مشروع Tasks API النهائي

المراجعة السريعة وهيكل المشروع النهائي

مرحباً بك في الدرس العشرين والأخير من سلسلة "تعلم بايثون من الصفر حتى بناء API عملي باستخدام FastAPI". بعد رحلة طويلة من الأساسيات إلى بناء API متكامل، حان الوقت لتجميع كل ما تعلمناه في مشروع واحد متكامل. في هذا الدرس، سنقوم ببناء مشروع Tasks API النهائي، وهو API كامل لإدارة المهام يمكن استخدامه في تطبيقات حقيقية.

لماذا هذا المشروع النهائي؟

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

ما الذي سنبنيه بالضبط؟

سنبني API لإدارة المهام (Tasks) يدعم العمليات الأساسية التالية:

• إنشاء مهمة جديدة (Create)
• قراءة جميع المهام (Read All)
• قراءة مهمة محددة (Read One)
• تحديث مهمة موجودة (Update)
• حذف مهمة (Delete)

بالإضافة إلى ذلك، سنضيف ميزات متقدمة مثل البحث والتصفية، والتحقق الكامل من صحة البيانات.

هيكل المشروع النهائي

لنبدأ بتنظيم ملفات المشروع بطريقة احترافية. هذا الهيكل يتبع أفضل الممارسات في مشاريع FastAPI:

tasks_api_project/
│
├── main.py                 # نقطة الدخول الرئيسية للتطبيق
├── database.py             # إعداد الاتصال بقاعدة البيانات
├── models.py               # نماذج قاعدة البيانات (SQLAlchemy)
├── schemas.py              # نماذج Pydantic للتحقق من البيانات
├── crud.py                 # دوال العمليات على قاعدة البيانات
└── requirements.txt        # قائمة المكتبات المطلوبة

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

الخطوة 1: إعداد البيئة والمكتبات

أولاً، تأكد من أنك في بيئة افتراضية. ثم قم بإنشاء ملف requirements.txt بالمحتوى التالي:

fastapi==0.104.1
uvicorn==0.24.0
sqlalchemy==2.0.23
pydantic==2.5.2
python-dotenv==1.0.0

ثم قم بتثبيتها باستخدام الأمر:

pip install -r requirements.txt

الخطوة 2: إنشاء ملف database.py

هذا الملف مسؤول عن الاتصال بقاعدة البيانات SQLite:

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}  # ضروري لـ SQLite
)

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

# الفئة الأساسية للنماذج
Base = declarative_base()

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

الخطوة 3: إنشاء نماذج قاعدة البيانات (models.py)

هنا نحدد هيكل جدول المهام في قاعدة البيانات:

from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.sql import func
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)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

الخطوة 4: إنشاء نماذج Pydantic (schemas.py)

هذه النماذج تستخدم للتحقق من صحة البيانات المدخلة والمخرجة:

from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

# النموذج الأساسي للمهمة
class TaskBase(BaseModel):
    title: str = Field(..., min_length=1, max_length=100)
    description: Optional[str] = Field(None, max_length=500)
    completed: bool = False

# نموذج إنشاء مهمة جديدة
class TaskCreate(TaskBase):
    pass

# نموذج تحديث مهمة
class TaskUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=100)
    description: Optional[str] = Field(None, max_length=500)
    completed: Optional[bool] = None

# نموذج عرض المهمة (يحتوي على جميع الحقول)
class TaskResponse(TaskBase):
    id: int
    created_at: datetime
    updated_at: Optional[datetime] = None

    class Config:
        from_attributes = True

الخطوة 5: إنشاء دوال CRUD (crud.py)

هذا الملف يحتوي على جميع العمليات التي نريد تنفيذها على قاعدة البيانات:

from sqlalchemy.orm import Session
from models import Task
from schemas import TaskCreate, TaskUpdate

def get_tasks(db: Session, skip: int = 0, limit: int = 100):
    """الحصول على قائمة المهام مع دعم التقسيم (pagination)"""
    return db.query(Task).offset(skip).limit(limit).all()

def get_task(db: Session, task_id: int):
    """الحصول على مهمة محددة باستخدام المعرف"""
    return db.query(Task).filter(Task.id == task_id).first()

def create_task(db: Session, task: TaskCreate):
    """إنشاء مهمة جديدة"""
    db_task = Task(**task.model_dump())
    db.add(db_task)
    db.commit()
    db.refresh(db_task)
    return db_task

def update_task(db: Session, task_id: int, task: TaskUpdate):
    """تحديث مهمة موجودة"""
    db_task = get_task(db, task_id)
    if db_task:
        update_data = task.model_dump(exclude_unset=True)
        for key, value in update_data.items():
            setattr(db_task, key, value)
        db.commit()
        db.refresh(db_task)
    return db_task

def delete_task(db: Session, task_id: int):
    """حذف مهمة"""
    db_task = get_task(db, task_id)
    if db_task:
        db.delete(db_task)
        db.commit()
    return db_task

الخطوة 6: إنشاء التطبيق الرئيسي (main.py)

هذا هو الملف الذي يربط كل شيء معاً:

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List

from database import engine, get_db, Base
from models import Task
from schemas import TaskCreate, TaskUpdate, TaskResponse
from crud import get_tasks, get_task, create_task, update_task, delete_task

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

# إنشاء تطبيق FastAPI
app = FastAPI(
    title="Tasks API",
    description="API كامل لإدارة المهام",
    version="1.0.0"
)

# نقطة النهاية الرئيسية
@app.get("/")
def read_root():
    return {"message": "مرحباً بك في Tasks API!"}

# الحصول على جميع المهام
@app.get("/tasks", response_model=List[TaskResponse])
def read_tasks(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    tasks = get_tasks(db, skip=skip, limit=limit)
    return tasks

# الحصول على مهمة محددة
@app.get("/tasks/{task_id}", response_model=TaskResponse)
def read_task(task_id: int, db: Session = Depends(get_db)):
    db_task = get_task(db, task_id=task_id)
    if db_task is None:
        raise HTTPException(status_code=404, detail="المهمة غير موجودة")
    return db_task

# إنشاء مهمة جديدة
@app.post("/tasks", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
def create_new_task(task: TaskCreate, db: Session = Depends(get_db)):
    return create_task(db=db, task=task)

# تحديث مهمة
@app.put("/tasks/{task_id}", response_model=TaskResponse)
def update_existing_task(task_id: int, task: TaskUpdate, db: Session = Depends(get_db)):
    db_task = update_task(db=db, task_id=task_id, task=task)
    if db_task is None:
        raise HTTPException(status_code=404, detail="المهمة غير موجودة")
    return db_task

# حذف مهمة
@app.delete("/tasks/{task_id}", response_model=TaskResponse)
def delete_existing_task(task_id: int, db: Session = Depends(get_db)):
    db_task = delete_task(db=db, task_id=task_id)
    if db_task is None:
        raise HTTPException(status_code=404, detail="المهمة غير موجودة")
    return db_task

تشغيل المشروع واختباره

لتشغيل المشروع، استخدم الأمر التالي في terminal:

uvicorn main:app --reload

ثم افتح المتصفح على http://127.0.0.1:8000/docs لترى واجهة Swagger الجميلة وتختبر جميع endpoints.

الأخطاء الشائعة وكيفية تجنبها

• خطأ "Table already exists": يحدث عند تشغيل التطبيق عدة مرات. الحل: استخدم Base.metadata.create_all(bind=engine) مرة واحدة فقط.
• خطأ 422 Validation Error: يعني أن البيانات التي أرسلتها لا تتطابق مع النموذج. تأكد من إرسال الحقول المطلوبة فقط.
• خطأ 404 Not Found: يحدث عندما تحاول الوصول إلى مهمة غير موجودة. تأكد من أن المعرف (ID) صحيح.

التمرين النهائي

للتأكد من فهمك الكامل، قم بالتالي:

1. أضف endpoint جديد يسمح بالبحث عن المهام حسب عنوانها (search by title).
2. أضف حقل "priority" للمهمة (عالي، متوسط، منخفض) وقم بتحديث جميع النماذج والدوال.
3. اختبر كل شيء باستخدام Swagger وتأكد من أن كل شيء يعمل بشكل صحيح.

مبروك! لقد أكملت السلسلة بأكملها وبنيت أول API عملي لك. الآن لديك أساس قوي لبناء أي API تريده في المستقبل. استمر في التعلم والتطوير!