تصميم RESTful ومصادقة JWT

في هذه المرحلة نربط الواجهة العامة لـ TaskFlow: نقاط نهاية API تتبع اصطلاحات REST، مؤمّنة بتوكنات JWT ومحمية بالتحكم بالوصول حسب الأدوار.

هيكل عناوين URL في REST

واجهة REST نظيفة تربط الموارد بعناوين URL وتستخدم طرق HTTP للتعبير عن القصد:

الطريقة	نمط العنوان	الغرض	رمز الحالة
POST	/api/v1/projects	إنشاء مشروع	201 Created
GET	/api/v1/projects	سرد المشاريع	200 OK
GET	/api/v1/projects/{id}	الحصول على مشروع واحد	200 OK
PUT	/api/v1/projects/{id}	تحديث مشروع	200 OK
DELETE	/api/v1/projects/{id}	حذف مشروع	204 No Content
POST	/api/v1/projects/{id}/tasks	إنشاء مهمة في مشروع	201 Created

قواعد عامة:

• استخدم أسماء الجمع للمجموعات (/projects وليس /project).
• ضع الموارد الفرعية تحت الأصلية (/projects/{id}/tasks).
• أضف إصدارًا لواجهتك البرمجية (/api/v1/) حتى لا تفاجئ التغييرات الجذرية العملاء.
• أرجع رمز الحالة الصحيح: 201 للإنشاء، 204 للحذف، 404 عند عدم الوجود، 403 عند الحظر، 422 لأخطاء التحقق.

تدفق مصادقة JWT

يستخدم TaskFlow توكنات JSON Web Tokens (JWT) للمصادقة عديمة الحالة:

التسجيل        تسجيل الدخول        الوصول لمسار محمي
────────►  ────────────►  ────────────────────────────────►
POST         POST              GET /api/v1/projects
/auth/       /auth/login       Authorization: Bearer <token>
register     ─► توكن JWT       ─► الخادم يفك تشفير التوكن
                                  ─► يحدد المستخدم
                                  ─► يرجع البيانات

يتكون JWT من ثلاثة أجزاء مفصولة بنقاط: header.payload.signature. يوقّع الخادم التوكن بمفتاح سري؛ وفي كل طلب يتحقق من التوقيع بدون الرجوع لقاعدة البيانات.

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

SECRET_KEY = "your-secret-key"  # يُحمّل من متغير البيئة
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

تجزئة كلمات المرور مع pwdlib + Argon2

توثيق FastAPI الحالي يوصي بـ pwdlib مع محرك Argon2 بدلاً من تركيبة passlib/bcrypt القديمة. Argon2 هو الفائز في مسابقة تجزئة كلمات المرور ومقاوم لهجمات القوة الغاشمة المعتمدة على GPU:

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

password_hash = PasswordHash((Argon2Hasher(),))

hashed = password_hash.hash("user-password")           # تجزئة
is_valid = password_hash.verify("user-password", hashed) # تحقق

حقن التبعيات في FastAPI للمصادقة

نظام Depends() في FastAPI يتيح لك حقن المستخدم الحالي في أي نقطة نهاية. التبعية تقرأ ترويسة Authorization: Bearer <token>، تفك تشفير JWT، تبحث عن المستخدم في قاعدة البيانات، وترجع كائن المستخدم (أو ترفع خطأ 401 Unauthorized):

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:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    except JWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

    user = await db.get(User, int(user_id))
    if user is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    return user

أي نقطة نهاية تحتاج مصادقة تضيف ببساطة current_user: User = Depends(get_current_user) لتوقيعها. FastAPI يتكفل بالباقي.

التحكم بالوصول حسب الأدوار (RBAC)

في TaskFlow، كل مستخدم له دور داخل المشروع (مالك أو مدير أو عضو). فحوصات الوصول تتم من خلال تبعية أخرى:

from enum import Enum

class ProjectRole(str, Enum):
    OWNER = "owner"
    ADMIN = "admin"
    MEMBER = "member"

def require_project_role(*allowed_roles: ProjectRole):
    async def checker(
        project_id: int,
        current_user: User = Depends(get_current_user),
        db: AsyncSession = Depends(get_db),
    ) -> ProjectMember:
        member = await get_project_member(db, project_id, current_user.id)
        if member is None or member.role not in allowed_roles:
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
        return member
    return checker

الاستخدام في نقطة نهاية:

@router.put("/projects/{project_id}")
async def update_project(
    project_id: int,
    data: ProjectUpdate,
    member: ProjectMember = Depends(
        require_project_role(ProjectRole.OWNER, ProjectRole.ADMIN)
    ),
    db: AsyncSession = Depends(get_db),
):
    # فقط المالك أو المدير يصل لهذا السطر
    ...

نمط التصفح بالصفحات

لنقاط نهاية القوائم، يستخدم TaskFlow تصفح page/size مع العدد الإجمالي لتمكين الواجهة الأمامية من عرض أدوات التنقل:

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

معلمات الاستعلام ?page=1&size=20 تحرك حساب الإزاحة: offset = (page - 1) * size. الاستجابة تتضمن دائمًا total (العدد الكامل) و pages (العدد الإجمالي للصفحات).

توثيق OpenAPI التلقائي

FastAPI يولّد توثيق API تفاعلي من الكود بدون جهد إضافي:

• Swagger UI على /docs -- اختبر نقاط النهاية مباشرة في المتصفح.
• ReDoc على /redoc -- مرجع نظيف وسهل القراءة.

نماذج Pydantic v2 ورموز حالة الاستجابة والمصادقة المحقونة بالتبعيات تظهر جميعها تلقائيًا في التوثيق المولّد. يمكن للعملاء تصدير JSON الخاص بـ OpenAPI من /openapi.json لتوليد SDKs بأي لغة.

---

التالي: معمل عملي -- ستبني جميع نقاط نهاية المصادقة والمشاريع والمهام لـ TaskFlow. :::