بناء نقاط نهاية API والمصادقة

الهدف

بناء طبقة REST API الكاملة لـ TaskFlow: نقاط نهاية المصادقة وعمليات CRUD للمشاريع وإدارة المهام. يجب أن تستخدم كل نقطة نهاية SQLAlchemy غير المتزامن ونماذج طلب/استجابة Pydantic v2 ورموز حالة HTTP صحيحة والتخويل المبني على JWT.

في نهاية هذا المعمل سيكون لديك API عامل مع 12 نقطة نهاية عبر ثلاثة موجهات.

المتطلبات الأساسية

يجب أن يكون لديك نماذج قاعدة البيانات والجلسة غير المتزامنة من المرحلة 2. هيكل مشروعك يجب أن يبدو هكذا:

taskflow/
├── app/
│   ├── core/
│   │   ├── config.py          # الإعدادات (SECRET_KEY, DATABASE_URL, إلخ)
│   │   ├── database.py        # المحرك غير المتزامن + تبعية get_db
│   │   └── security.py        # ← جديد: JWT + تجزئة كلمات المرور
│   ├── models/
│   │   ├── user.py
│   │   ├── project.py
│   │   └── task.py
│   ├── schemas/               # ← جديد: نماذج Pydantic v2
│   │   ├── auth.py
│   │   ├── project.py
│   │   ├── task.py
│   │   └── common.py
│   ├── api/                   # ← جديد: معالجات المسارات
│   │   ├── deps.py            # get_current_user, require_project_role
│   │   ├── auth.py            # /api/v1/auth/*
│   │   ├── projects.py        # /api/v1/projects/*
│   │   └── tasks.py           # /api/v1/projects/{id}/tasks/* و /api/v1/tasks/*
│   └── main.py                # تطبيق FastAPI مع تضمين الموجهات
├── requirements.txt
└── .env

الجزء 1 — أدوات الأمان (app/core/security.py)

أنشئ أدوات تجزئة كلمات المرور وتوكنات JWT.

تجزئة كلمات المرور — استخدم pwdlib مع Argon2:

from pwdlib import PasswordHash
from pwdlib.hashers.argon2 import Argon2Hasher

password_hash = PasswordHash((Argon2Hasher(),))

def hash_password(password: str) -> str:
    return password_hash.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return password_hash.verify(plain_password, hashed_password)

توكنات JWT — استخدم python-jose[cryptography]:

from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError

def create_access_token(subject: str, expires_delta: timedelta | None = None) -> str:
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=30))
    to_encode = {"sub": subject, "exp": expire}
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")

الجزء 2 — مخططات Pydantic v2 (app/schemas/)

أنشئ نماذج الطلب والاستجابة. جميع النماذج يجب أن تستخدم صيغة Pydantic v2 (model_config = ConfigDict(from_attributes=True)).

app/schemas/common.py — استجابة التصفح المشتركة:

from pydantic import BaseModel
from typing import Generic, TypeVar, Sequence

T = TypeVar("T")

class PaginatedResponse(BaseModel, Generic[T]):
    items: Sequence[T]
    total: int
    page: int
    size: int
    pages: int

app/schemas/auth.py — نماذج طلب/استجابة المصادقة:

• UserCreate: email (EmailStr), password (8 أحرف كحد أدنى), full_name
• UserResponse: id, email, full_name, created_at — مع model_config = ConfigDict(from_attributes=True)
• Token: access_token, token_type

app/schemas/project.py — نماذج المشاريع:

• ProjectCreate: name (1-100 حرف), description (اختياري)
• ProjectUpdate: name (اختياري), description (اختياري)
• ProjectResponse: id, name, description, owner_id, created_at, updated_at

app/schemas/task.py — نماذج المهام:

• TaskCreate: title (1-200 حرف), description (اختياري), priority (low/medium/high/urgent), assigned_to (معرف مستخدم اختياري)
• TaskUpdate: title (اختياري), description (اختياري), status (todo/in_progress/done), priority (اختياري), assigned_to (اختياري)
• TaskResponse: id, title, description, status, priority, assigned_to, project_id, created_by, created_at, updated_at

الجزء 3 — تبعيات المصادقة (app/api/deps.py)

أنشئ تبعيات FastAPI التي تحمي نقاط النهاية.

get_current_user — يفك تشفير توكن JWT Bearer ويرجع المستخدم من قاعدة البيانات. يرفع خطأ 401 Unauthorized إذا كان التوكن مفقودًا أو منتهيًا أو المستخدم غير موجود.

require_project_role(*allowed_roles) — يرجع تبعية تفحص دور المستخدم الحالي في مشروع معين. يرفع خطأ 403 Forbidden إذا لم يكن المستخدم عضوًا أو دوره غير مسموح.

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = await db.get(User, int(user_id))
    if user is None:
        raise credentials_exception
    return user

الجزء 4 — موجه المصادقة (app/api/auth.py)

نفّذ ثلاث نقاط نهاية:

POST /api/v1/auth/register

• جسم الطلب: UserCreate
• تحقق إذا كان البريد الإلكتروني موجودًا → 409 Conflict
• جزّئ كلمة المرور، أنشئ المستخدم في قاعدة البيانات
• أرجع UserResponse مع حالة 201

POST /api/v1/auth/login

• اقبل OAuth2PasswordRequestForm (username = email, password)
• تحقق من وجود البريد الإلكتروني وتطابق كلمة المرور → 401 Unauthorized إن لم يتطابق
• أرجع Token مع توكن JWT

GET /api/v1/auth/me

• يتطلب مصادقة (تبعية get_current_user)
• أرجع UserResponse للمستخدم الحالي

الجزء 5 — موجه المشاريع (app/api/projects.py)

نفّذ أربع نقاط نهاية:

POST /api/v1/projects

• يتطلب مصادقة
• جسم الطلب: ProjectCreate
• أنشئ المشروع مع owner_id = current_user.id
• أنشئ سجل ProjectMember بدور owner
• أرجع ProjectResponse مع حالة 201

GET /api/v1/projects

• يتطلب مصادقة
• معلمات الاستعلام: page (افتراضي 1)، size (افتراضي 20، أقصى 100)
• أرجع فقط المشاريع التي المستخدم الحالي عضو فيها
• أرجع PaginatedResponse[ProjectResponse]

GET /api/v1/projects/{project_id}

• يتطلب مصادقة + عضوية في المشروع (أي دور)
• أرجع ProjectResponse
• أرجع 404 إذا لم يوجد، 403 إذا لم يكن عضوًا

PUT /api/v1/projects/{project_id}

• يتطلب مصادقة + دور في المشروع: مالك أو مدير
• جسم الطلب: ProjectUpdate
• أرجع ProjectResponse المحدّث

DELETE /api/v1/projects/{project_id}

• يتطلب مصادقة + دور في المشروع: مالك فقط
• احذف المشروع وجميع المهام/الأعضاء المرتبطين
• أرجع 204 No Content

الجزء 6 — موجه المهام (app/api/tasks.py)

نفّذ ثلاث نقاط نهاية:

POST /api/v1/projects/{project_id}/tasks

• يتطلب مصادقة + عضوية في المشروع (أي دور)
• جسم الطلب: TaskCreate
• عيّن created_by = current_user.id، project_id = project_id، status = "todo"
• أرجع TaskResponse مع حالة 201

GET /api/v1/projects/{project_id}/tasks

• يتطلب مصادقة + عضوية في المشروع (أي دور)
• معلمات الاستعلام: page، size، status (تصفية)، priority (تصفية)، assigned_to (تصفية)
• ابنِ استعلام SQLAlchemy ديناميكي مع مرشحات اختيارية
• أرجع PaginatedResponse[TaskResponse]

مثال على الطلب:

GET /api/v1/projects/1/tasks?page=1&size=20&status=todo&priority=high

PUT /api/v1/tasks/{task_id}

• يتطلب مصادقة + عضوية في مشروع المهمة (أي دور)
• ابحث عن المهمة بالمعرف → 404 إذا لم توجد
• تحقق أن المستخدم عضو في مشروع المهمة → 403 إذا لم يكن
• جسم الطلب: TaskUpdate
• أرجع TaskResponse المحدّث

DELETE /api/v1/tasks/{task_id}

• يتطلب مصادقة + دور في مشروع المهمة: مدير أو مالك
• أرجع 204 No Content

الجزء 7 — ربط التطبيق (app/main.py)

سجّل جميع الموجهات الثلاثة تحت البادئة /api/v1:

from fastapi import FastAPI
from app.api import auth, projects, tasks

app = FastAPI(title="TaskFlow API", version="1.0.0")

app.include_router(auth.router, prefix="/api/v1/auth", tags=["auth"])
app.include_router(projects.router, prefix="/api/v1/projects", tags=["projects"])
app.include_router(tasks.router, prefix="/api/v1", tags=["tasks"])

الجزء 8 — الاختبار مع Swagger UI

شغّل الخادم وافتح http://localhost:8000/docs. نفّذ هذا التسلسل:

1. سجّل مستخدمًا عبر POST /api/v1/auth/register
2. سجّل الدخول عبر POST /api/v1/auth/login — انسخ access_token
3. انقر على Authorize في Swagger UI والصق التوكن
4. أنشئ مشروعًا عبر POST /api/v1/projects
5. اسرد المشاريع عبر GET /api/v1/projects
6. أنشئ مهمة عبر POST /api/v1/projects/{id}/tasks
7. اسرد المهام مع المرشحات عبر GET /api/v1/projects/{id}/tasks?status=todo

ما يجب تقديمه

يجب أن يحتوي تقديمك على 10 أقسام ملفات في المحرر أدناه. يبدأ كل قسم بعنوان # FILE N:.

---

تلميحات

• استخدم select().where(...).offset(...).limit(...) للاستعلامات المصفّحة
• استخدم func.count() للحصول على العدد الإجمالي بكفاءة
• للتصفية، اربط استدعاءات .where() بشكل شرطي:

  query = select(Task).where(Task.project_id == project_id)
  if status:
      query = query.where(Task.status == status)
  if priority:
      query = query.where(Task.priority == priority)
  

• استخدم math.ceil(total / size) لحساب عدد الصفحات
• تذكر تأكيد الجلسة بعد عمليات الإنشاء/التحديث/الحذف