بناء نماذج قاعدة بيانات TaskFlow والترحيلات
التعليمات
بناء طبقة قاعدة بيانات TaskFlow
أنت تبني طبقة قاعدة البيانات الكاملة لـ TaskFlow -- واجهة REST API إنتاجية لإدارة المهام. بنهاية هذا المختبر سيكون لديك نماذج SQLAlchemy غير متزامنة ومخططات Pydantic للتحقق وترحيلات Alembic وسكريبت بذر لقاعدة البيانات.
الأدوات: SQLAlchemy 2.0.46 | asyncpg | Alembic 1.18.4 | PostgreSQL 18.2 | Pydantic 2.12.x
الجزء 1: جلسة قاعدة بيانات غير متزامنة (15 نقطة)
أنشئ app/database.py مع محرك قاعدة البيانات غير المتزامن ومصنع الجلسات.
المتطلبات:
- استخدم
create_async_engineمع مشغّلpostgresql+asyncpg - اقرأ رابط قاعدة البيانات من متغير البيئة
DATABASE_URL - أنشئ
async_sessionmakerمعexpire_on_commit=False - نفّذ مولّد غير متزامن
get_dbيُنتجAsyncSession - عرّف فئة
BaseباستخدامDeclarativeBaseلترث منها جميع النماذج
# الاستيرادات المتوقعة
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
الجزء 2: نماذج SQLAlchemy 2.0 (30 نقطة)
أنشئ app/models.py مع أربعة نماذج باستخدام التعليقات التوضيحية Mapped من SQLAlchemy 2.0.
نموذج User
| العمود | النوع | القيود |
|---|---|---|
| id | int | مفتاح أساسي |
| str(255) | فريد، مفهرس، غير فارغ | |
| hashed_password | str(255) | غير فارغ |
| full_name | str(100) | غير فارغ |
| is_active | bool | افتراضي True |
| created_at | datetime | افتراضي utcnow |
| updated_at | datetime | افتراضي utcnow، يُحدّث عند التغيير |
العلاقات: owned_projects، assigned_tasks، memberships
نموذج Project
| العمود | النوع | القيود |
|---|---|---|
| id | int | مفتاح أساسي |
| name | str(200) | غير فارغ |
| description | str(1000) | يقبل القيمة الفارغة |
| owner_id | int | مفتاح خارجي إلى users.id، غير فارغ |
| created_at | datetime | افتراضي utcnow |
| updated_at | datetime | افتراضي utcnow، يُحدّث عند التغيير |
العلاقات: owner، tasks، members
نموذج Task
| العمود | النوع | القيود |
|---|---|---|
| id | int | مفتاح أساسي |
| title | str(300) | غير فارغ |
| description | str(2000) | يقبل القيمة الفارغة |
| status | تعداد TaskStatus | افتراضي "todo"، غير فارغ |
| priority | تعداد TaskPriority | افتراضي "medium"، غير فارغ |
| project_id | int | مفتاح خارجي إلى projects.id، غير فارغ |
| assignee_id | int | مفتاح خارجي إلى users.id، يقبل القيمة الفارغة |
| due_date | datetime | يقبل القيمة الفارغة |
| created_at | datetime | افتراضي utcnow |
| updated_at | datetime | افتراضي utcnow، يُحدّث عند التغيير |
العلاقات: project، assignee
التعدادات:
TaskStatus:todo،in_progress،doneTaskPriority:low،medium،high
نموذج ProjectMember
| العمود | النوع | القيود |
|---|---|---|
| id | int | مفتاح أساسي |
| project_id | int | مفتاح خارجي إلى projects.id، غير فارغ |
| user_id | int | مفتاح خارجي إلى users.id، غير فارغ |
| role | تعداد MemberRole | افتراضي "member"، غير فارغ |
| joined_at | datetime | افتراضي utcnow |
التعداد:
MemberRole:owner،admin،member
القيد: فريد معًا على (project_id, user_id) -- يمكن للمستخدم أن يكون له دور واحد فقط لكل مشروع.
جميع النماذج يجب أن:
- تستخدم
Mapped[type]وmapped_column()(أسلوب SQLAlchemy 2.0) - ترث من
Baseالمشترك - تعرّف
__tablename__بشكل صريح - تستخدم
relationship()معback_populatesللروابط ثنائية الاتجاه
الجزء 3: مخططات Pydantic v2 (25 نقطة)
أنشئ app/schemas.py مع مخططات الإنشاء والتحديث والاستجابة لكل نموذج.
المتطلبات:
- جميع مخططات الاستجابة يجب أن تضبط
model_config = ConfigDict(from_attributes=True) - استخدم
EmailStrللتحقق من البريد الإلكتروني فيUserCreate - مخططات التحديث يجب أن تجعل جميع الحقول
Optional(تحديثات جزئية) - استخدم
enum.Enumمن Python لحقول الحالة/الأولوية/الدور - أضف قيود الحقول حيثما كان مناسبًا (مثل
min_length=1للعنوان)
المخططات المتوقعة:
# مخططات المستخدم
class UserCreate(BaseModel): # email, password, full_name
class UserUpdate(BaseModel): # full_name (اختياري), is_active (اختياري)
class UserResponse(BaseModel): # id, email, full_name, is_active, created_at, updated_at
# مخططات المشروع
class ProjectCreate(BaseModel): # name, description (اختياري)
class ProjectUpdate(BaseModel): # name (اختياري), description (اختياري)
class ProjectResponse(BaseModel): # id, name, description, owner_id, created_at, updated_at
# مخططات المهمة
class TaskCreate(BaseModel): # title, description (اختياري), status, priority, project_id, assignee_id (اختياري), due_date (اختياري)
class TaskUpdate(BaseModel): # جميع الحقول اختيارية
class TaskResponse(BaseModel): # جميع الحقول بما فيها id, created_at, updated_at
# مخططات عضو المشروع
class ProjectMemberCreate(BaseModel): # user_id, role
class ProjectMemberResponse(BaseModel): # id, project_id, user_id, role, joined_at
الجزء 4: إعداد Alembic والترحيل الأولي (15 نقطة)
أعدّ Alembic للترحيلات غير المتزامنة وولّد الترحيل الأولي.
الخطوات:
- هيّئ Alembic بـ
alembic init alembic - اضبط
alembic/env.pyللعمل غير المتزامن:- استورد
Base.metadataكـtarget_metadata - استخدم
run_asyncمعconnectable = create_async_engine() - اضبط
context.configure()معrender_as_batch=Trueلتوافق SQLite أثناء الاختبارات
- استورد
- حدّث
alembic.iniبالرابط الصحيحsqlalchemy.url - ولّد الترحيل الأولي:
alembic revision --autogenerate -m "create users projects tasks project_members tables" - الترحيل المولّد يجب أن يتضمن دوال
upgrade()وdowngrade()التي تنشئ وتحذف جميع الجداول الأربعة
قدّم ملف alembic/env.py الكامل الذي يظهر التكوين غير المتزامن.
الجزء 5: سكريبت بذر قاعدة البيانات (15 نقطة)
أنشئ app/seed.py الذي يملأ قاعدة البيانات ببيانات اختبارية واقعية.
المتطلبات:
- أنشئ 3 مستخدمين على الأقل مع كلمات مرور مُجزّأة (استخدم تجزئة بديلة)
- أنشئ مشروعين على الأقل، كل منهما مملوك لمستخدم مختلف
- أنشئ 5 مهام على الأقل عبر المشاريع بحالات وأولويات متنوعة
- أضف أعضاء مشروع بأدوار مختلفة
- استخدم الجلسة غير المتزامنة لإدراج جميع البيانات
- يجب أن يكون السكريبت قابلاً للتشغيل بـ
python -m app.seed - ضمّن معالجة أخطاء صحيحة (تخطّى إذا كانت البيانات موجودة مسبقًا)
# الهيكل المتوقع
async def seed_database():
async with async_session() as session:
# تحقق مما إذا كانت البيانات موجودة مسبقًا
# أنشئ المستخدمين
# أنشئ المشاريع
# أنشئ أعضاء المشروع
# أنشئ المهام
await session.commit()
if __name__ == "__main__":
import asyncio
asyncio.run(seed_database())
ما يجب تقديمه
يجب أن يحتوي تقديمك على 5 أقسام ملفات في المحرر أدناه. يبدأ كل قسم بعنوان # FILE N:.