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

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

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

قدّم SQLAlchemy 2.0 دعمًا أصليًا للعمليات غير المتزامنة عبر create_async_engine و AsyncSession. بالاقتران مع مشغّل asyncpg لـ PostgreSQL، يمنحك هذا وصولًا غير حاجب لقاعدة البيانات يتوافق مع بنية 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	يربط المستخدم بالمشروع مع دور

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

قراران في النمذجة يستحقان التصريح بهما، لأنهما ما يدور حوله الجدل في المراجعات:

‏assignee_id يقبل القيمة الفارغة، أما project_id فلا. المهمة بلا مُسنَد إليه حالة طبيعية — إنها قائمة الانتظار. أما المهمة بلا مشروع فليست حالة أصلاً، بل بيانات تالفة. قابلية الفراغ هي المكان الذي تُرمّز فيه أيّاً من الحالتين تعتبرها مشروعة.

الدور يعيش على ProjectMember لا على User. وضع عمود role على User يجعل الأدوار عامة، فيصبح معنى «مدير» مديراً لكل شيء. ولأن الدور يجلس على صف الربط، يمكن للشخص نفسه أن يملك مشروعاً ويكون عضواً بصلاحية قراءة فقط في آخر — وهذا ما يتوقعه الناس فعلاً من أداة مساحات عمل.

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

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

from datetime import datetime, timezone
from sqlalchemy import String, DateTime, ForeignKey, func, 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(
        DateTime(timezone=True), server_default=func.now()
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=lambda: datetime.now(timezone.utc),
    )

    # العلاقات
    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")

تفصيلان في أعمدة الوقت أعلاه مقصودان، وكلاهما مما يؤذي في الإنتاج.

‏datetime.utcnow غائبة عن هذا الكود عن قصد. فهي مهملة منذ Python 3.12، والسبب يستحق الفهم لا الحفظ: تُرجع كائن datetime ساذجاً — بلا منطقة زمنية — يصادف أنه يحمل توقيت UTC. وكل مقارنة لاحقة عليها أن تتذكر حقيقة لا يسجلها الكائن نفسه، وفي النهاية تنسى إحداها. البديل هو datetime.now(timezone.utc) التي تُرجع كائناً واعياً بالمنطقة الزمنية يستحيل إساءة قراءته. لاحظ أنها تأخذ وسيطاً، فكقيمة افتراضية لعمود تحتاج إلى تغليفها بدالة: lambda: datetime.now(timezone.utc).

‏server_default=func.now() أفضل من قيمة افتراضية من جانب Python لطوابع الإنشاء. فهي تجعل PostgreSQL نفسه يختم الصف، فتحصل الصفوف المُدرَجة عبر ترحيل أو سكربت بذر أو جلسة psql على طابع زمني أيضاً — لا الصفوف التي مرّت صدفةً عبر ORM فقط. واقرنها بـ DateTime(timezone=True) ليكون العمود من نوع timestamptz فتحفظ قاعدة البيانات الإزاحة بدل أن تهملها.

التحول الآخر الجدير بالملاحظة بنيوي: Mapped[int] وmapped_column يستبدلان أسلوب Column(Integer) القديم، وهذا ما يمنحك دعم مدقق الأنواع.

1from sqlalchemy import Column, Integer, String, Boolean
2from sqlalchemy.orm import declarative_base, relationship
3 
4Base = declarative_base()
5 
6class User(Base):
7    __tablename__ = "users"
8 
9    id = Column(Integer, primary_key=True)
10    email = Column(String(255), unique=True, index=True)
11    is_active = Column(Boolean, default=True)
12 
13    # مدقق الأنواع يرى: Column لا int
14    # user.id + 1  -> بلا تحذير وبلا مساعدة
15    projects = relationship("Project", back_populates="owner")

1from sqlalchemy import String
2from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
3 
4class Base(DeclarativeBase):
5    pass
6 
7class User(Base):
8    __tablename__ = "users"
9 
10    id: Mapped[int] = mapped_column(primary_key=True)
11    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
12    is_active: Mapped[bool] = mapped_column(default=True)
13 
14    # مدقق الأنواع يرى: int و str و bool
15    # قابلية الفراغ تأتي من Mapped[str] مقابل Mapped[str | None]
16    projects: Mapped[list["Project"]] = relationship(back_populates="owner")

التعليق التوضيحي ليس زينة. في أسلوب 2.0، Mapped[str] وMapped[str | None] هما ما يعلن العمود NOT NULL أو قابلاً للفراغ — فيُشتق مدقق الأنواع ومخطط قاعدة البيانات من العبارة نفسها ويستحيل أن يختلفا. أما في الأسلوب القديم فكانا إعلانين منفصلين، وإبقاؤهما متوافقين كان عملاً يدوياً لم يتقنه أحد قط.

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

‏Alembic هو أداة الترحيل المبنية لـ SQLAlchemy. فكّر فيه كـ "git لمخطط قاعدة بياناتك" — مع فارق واحد مهم عن git، تُعلّمه البطاقة الكهرمانية في المخطط أدناه.

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

الخطوة الكهرمانية هي التي يجب استيعابها. --autogenerate أداة مقارنة لا قارئ أفكار: إعادة تسمية full_name إلى display_name تُنتج drop_column يتبعه add_column، وهو ما ينجح في كل اختبار على قاعدة بيانات تطوير فارغة ويمحو بصمت محتوى عمود في الإنتاج. اقرأ الملف المُولَّد في كل مرة، وأعد كتابة هذا الثنائي كـ op.alter_column(..., new_column_name=...) إن كان مقصودك إعادة تسمية.

مخططات Pydantic v2

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

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. كتابة سكريبت بذر لملء قاعدة البيانات ببيانات اختبارية

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