تصميم قاعدة بيانات إنتاجية مع SQLAlchemy 2.0

في المرحلة الأولى أعددت هيكل المشروع. الآن حان الوقت لبناء أساس البيانات. كل واجهة API إنتاجية تحتاج طبقة قاعدة بيانات متينة -- ولـ TaskFlow، هذا يعني نماذج SQLAlchemy غير متزامنة ومخططات Pydantic للتحقق وترحيلات مُتحكم بإصداراتها مع Alembic.

لماذا SQLAlchemy غير المتزامن؟

قدّم SQLAlchemy 2.0 (الإصدار الحالي: 2.0.46، يناير 2026) دعمًا أصليًا للعمليات غير المتزامنة عبر create_async_engine و AsyncSession. بالاقتران مع مشغّل asyncpg لـ PostgreSQL 18، يمنحك هذا وصولًا غير حاجب لقاعدة البيانات يتوافق مع بنية FastAPI غير المتزامنة.

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/taskflow"

engine = create_async_engine(DATABASE_URL, echo=False, pool_size=20)
async_session = async_sessionmaker(engine, expire_on_commit=False)

async def get_db():
    async with async_session() as session:
        yield session

النقاط الرئيسية:

• create_async_engine يستبدل create_engine المتزامن
• async_sessionmaker (وليس sessionmaker القديم) ينشئ جلسات متوافقة مع البرمجة غير المتزامنة
• expire_on_commit=False يمنع أخطاء التحميل الكسول بعد الحفظ في السياقات غير المتزامنة
• pool_size=20 نقطة بداية معقولة للإنتاج لتجمع الاتصالات

مخطط قاعدة بيانات TaskFlow

يحتاج TaskFlow إلى أربعة جداول أساسية بعلاقات واضحة:

الجدول	الغرض	العلاقات الرئيسية
User	الحسابات مع بيانات المصادقة	يملك المشاريع، يُسند إليه المهام
Project	حاويات المهام	يحتوي مهام متعددة، له أعضاء
Task	عناصر العمل الفردية	ينتمي لمشروع، يُسند لمستخدم
ProjectMember	جدول ربط RBAC	يربط المستخدم بالمشروع مع دور

علاقات الكيانات تبدو هكذا:

User 1──* Project        (owner_id)
User 1──* Task           (assignee_id, nullable)
User 1──* ProjectMember  (user_id)
Project 1──* Task        (project_id)
Project 1──* ProjectMember (project_id)

تعريف النماذج مع SQLAlchemy 2.0

يستخدم SQLAlchemy 2.0 التعليقات التوضيحية DeclarativeBase و Mapped. هذا تحوّل كبير عن أسلوب declarative_base() القديم:

from datetime import datetime
from sqlalchemy import String, ForeignKey, Enum as SAEnum
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
import enum

class Base(DeclarativeBase):
    pass

class TaskStatus(str, enum.Enum):
    todo = "todo"
    in_progress = "in_progress"
    done = "done"

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    full_name: Mapped[str] = mapped_column(String(100))
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    updated_at: Mapped[datetime] = mapped_column(
        default=datetime.utcnow, onupdate=datetime.utcnow
    )

    # العلاقات
    owned_projects: Mapped[list["Project"]] = relationship(back_populates="owner")
    assigned_tasks: Mapped[list["Task"]] = relationship(back_populates="assignee")
    memberships: Mapped[list["ProjectMember"]] = relationship(back_populates="user")

لاحظ كيف أن Mapped[int] و mapped_column يستبدلان أسلوب Column(Integer) القديم. هذا يمنحك دعمًا كاملًا لمدقق الأنواع.

لماذا Alembic للترحيلات

Alembic (الإصدار الحالي: 1.18.4، فبراير 2026) هو أداة الترحيل المبنية لـ SQLAlchemy. فكّر فيه كـ "git لمخطط قاعدة بياناتك":

# تهيئة Alembic في مشروعك
alembic init alembic

# توليد ترحيل من تغييرات النماذج
alembic revision --autogenerate -m "create initial tables"

# تطبيق الترحيلات على قاعدة البيانات
alembic upgrade head

كل ترحيل هو ملف Python يحتوي دوال upgrade() و downgrade(). يمكن لفريقك مراجعة تغييرات المخطط في طلبات السحب، والتراجع بأمان، والحفاظ على تزامن كل البيئات.

مخططات Pydantic v2

يتولى Pydantic v2 (2.12.x) التحقق من الطلبات والاستجابات. تنشئ مخططات منفصلة لعمليات مختلفة:

from pydantic import BaseModel, EmailStr, ConfigDict
from datetime import datetime

class UserCreate(BaseModel):
    email: EmailStr
    password: str
    full_name: str

class UserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    email: str
    full_name: str
    is_active: bool
    created_at: datetime

ConfigDict(from_attributes=True) يستبدل orm_mode = True القديم من Pydantic v1. هذا يسمح لك بتمرير كائنات نماذج SQLAlchemy مباشرة إلى مخططات الاستجابة.

ما ستبنيه

في المختبر العملي التالي، ستقوم بـ:

1. تعريف جميع نماذج SQLAlchemy 2.0 الأربعة مع العلاقات والتعدادات المناسبة
2. إنشاء مصنع جلسات قاعدة بيانات غير متزامن
3. كتابة مخططات Pydantic v2 لمتغيرات الإنشاء والتحديث والاستجابة لكل نموذج
4. تهيئة Alembic وتوليد أول ترحيل
5. كتابة سكريبت بذر لملء قاعدة البيانات ببيانات اختبارية

التالي: بناء طبقة قاعدة البيانات الكاملة في المختبر العملي. :::