تزریق وابستگی، ساختار، خطاها و تنظیمات

۱. وابستگی: «FastAPI، این را برایم آماده کن»

دو endpoint به پارامترهای صفحه‌بندی یکسانی نیاز دارند. به جای تکرار، یک تابع می‌نویسیم و آن را به عنوان وابستگی اعلام می‌کنیم:

import uvicorn
from fastapi import Depends, FastAPI

app = FastAPI()

COURSES = [f'Course {n}' for n in range(1, 31)]
USERS = [f'user{n}@example.com' for n in range(1, 13)]


def pagination(page: int = 1, size: int = 10) -> dict[str, int]:
    size = min(size, 50)  # never more than 50 per page
    return {'offset': (page - 1) * size, 'limit': size}


@app.get('/courses')
def list_courses(paging: dict[str, int] = Depends(pagination)) -> list[str]:
    start = paging['offset']
    return COURSES[start : start + paging['limit']]


@app.get('/users')
def list_users(paging: dict[str, int] = Depends(pagination)) -> list[str]:
    start = paging['offset']
    return USERS[start : start + paging['limit']]


if __name__ == '__main__':
    uvicorn.run('app:app', reload=True)

یک تابع صفحه‌بندی برای دو endpoint: نمونه 01_depends_basics/app.py

در هر درخواست، FastAPI ابتدا pagination(...) را صدا می‌زند، پارامترهایش را از رشته پرس‌وجو می‌خواند و نتیجه را به endpoint می‌دهد. پارامترهای وابستگی در /docs طوری نشان داده می‌شوند که انگار مال خود endpoint هستند. این همان تابع مرتبه بالای فصل اول است: ما تابع را به FastAPI می‌دهیم و FastAPI تصمیم می‌گیرد کی صدایش بزند.

۲. شکل امروزی: Annotated

from dataclasses import dataclass
from typing import Annotated

import uvicorn
from fastapi import Depends, FastAPI

app = FastAPI()
calls = {'get_max_size': 0}


@dataclass
class PageParams:
    page: int
    size: int

    @property
    def offset(self) -> int:
        return (self.page - 1) * self.size


def get_max_size() -> int:
    calls['get_max_size'] += 1
    return 50


def get_paging(
    max_size: Annotated[int, Depends(get_max_size)], page: int = 1, size: int = 10
) -> PageParams:
    return PageParams(page=max(page, 1), size=min(size, max_size))


Paging = Annotated[PageParams, Depends(get_paging)]
MaxSize = Annotated[int, Depends(get_max_size)]


@app.get('/courses')
def list_courses(paging: Paging, max_size: MaxSize) -> dict:
    return {
        'offset': paging.offset,
        'size': paging.size,
        'max_size': max_size,
        'get_max_size was called': calls['get_max_size'],  # grows by 1 per request
    }


if __name__ == '__main__':
    uvicorn.run('app:app', reload=True)

نام مستعار Annotated، وابستگی تودرتو و کش: نمونه 02_annotated_dependencies/app.py

• Paging = Annotated[PageParams, Depends(get_paging)] یک بار نوشته می‌شود و همه‌جا استفاده می‌شود.
• یک وابستگی می‌تواند به وابستگی دیگری وابسته باشد. FastAPI کل درخت را حل می‌کند.
• در یک درخواست هر وابستگی فقط یک بار صدا زده می‌شود، حتی اگر چند جا به آن نیاز باشد. شمارنده خروجی این را نشان می‌دهد.

۳. کلاس و سرویس به عنوان وابستگی

هر چیز صدازدنی می‌تواند وابستگی باشد، از جمله یک کلاس: FastAPI پارامترهای __init__ آن را می‌خواند و Depends() بدون آرگومان یعنی «از خود نوع استفاده کن». کاربرد مهم‌تر، سرویس است: endpoint فقط سرویس را می‌خواهد و نمی‌داند سرویس چگونه ساخته می‌شود. امروز سرویس از یک فهرست استفاده می‌کند؛ در فصل پنجم از یک نشست پایگاه داده، و endpoint تغییری نمی‌کند. این وارونگی وابستگی در عمل است.

۴. وابستگی با yield: آماده کن، تحویل بده، همیشه جمع کن

این همان مدیر زمینه فصل اول است، به شکل وابستگی:

class FakeSession:
    counter = 0

    def __init__(self) -> None:
        FakeSession.counter += 1
        self.number = FakeSession.counter
        print(f'[session {self.number}] OPEN')

    def rollback(self) -> None:
        print(f'[session {self.number}] ROLLBACK')

    def close(self) -> None:
        print(f'[session {self.number}] CLOSE')


def get_session() -> Iterator[FakeSession]:
    session = FakeSession()
    try:
        yield session
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()


SessionDep = Annotated[FakeSession, Depends(get_session)]
app = FastAPI()


@app.get('/ok')
def ok(session: SessionDep) -> dict[str, int]:
    return {'session': session.number}


@app.get('/fail')
def fail(session: SessionDep) -> None:
    raise HTTPException(400, detail='something was wrong with the request')

باز کردن، بستن و بازگرداندن: نمونه 04_yield_dependencies/app.py

[session 1] OPEN
[session 1] CLOSE
[session 2] OPEN
[session 2] ROLLBACK
[session 2] CLOSE

در فصل پنجم دقیقاً همین شکل به هر درخواست نشست پایگاه داده خودش را می‌دهد.

۵. هدر، کوکی و اولین وابستگی احراز هویت

app = FastAPI()

API_KEYS = {'secret-admin-key': 'admin', 'secret-teacher-key': 'teacher'}


@app.get('/whoami')
def whoami(
    user_agent: Annotated[str | None, Header()] = None,
    accept_language: Annotated[str | None, Header()] = None,
) -> dict[str, str | None]:
    return {'user_agent': user_agent, 'language': accept_language}


@app.post('/preferences/theme/{theme}')
def set_theme(theme: str, response: Response) -> dict[str, str]:
    response.set_cookie('theme', theme, max_age=3600, httponly=True, samesite='lax')
    return {'saved': theme}


@app.get('/preferences')
def get_preferences(theme: Annotated[str | None, Cookie()] = None) -> dict[str, str]:
    return {'theme': theme or 'light (default)'}


def get_caller(x_api_key: Annotated[str | None, Header()] = None) -> str:
    if x_api_key not in API_KEYS:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED, detail='Missing or invalid X-API-Key header'
        )
    return API_KEYS[x_api_key]


@app.delete('/courses/{course_id}')
def delete_course(course_id: int, caller: Annotated[str, Depends(get_caller)]) -> dict:
    return {'deleted': course_id, 'by': caller}


if __name__ == '__main__':
    uvicorn.run('app:app', reload=True)

Header، Cookie و کلید API: نمونه 05_headers_and_cookies/app.py

• user_agent: Annotated[str | None, Header()] هدر User-Agent را می‌خواند. زیرخط در نام پایتونی به خط تیره تبدیل می‌شود.
• response.set_cookie(...) از مرورگر می‌خواهد کوکی را ذخیره کند و Cookie() آن را در درخواست بعدی می‌خواند.
• get_caller مفهوم احراز هویت است: وابستگی‌ای که یک هدر را می‌خواند و یا «چه کسی صدا زده» را برمی‌گرداند، یا با ۴۰۱ درخواست را متوقف می‌کند. فصل‌های نهم و دهم کلید ثابت را با کاربر واقعی و توکن JWT عوض می‌کنند، اما شکل همین می‌ماند.

۶. مدیریت متمرکز خطا

سرویس فصل اول NotFoundError و ConflictError پرتاب می‌کرد. سرویس نباید از HTTP خبر داشته باشد، پس به FastAPI یاد می‌دهیم چطور ترجمه کند:

class AppError(Exception):
    status_code = 400
    code = 'app_error'

    def __init__(self, message: str) -> None:
        super().__init__(message)
        self.message = message


class NotFoundError(AppError):
    status_code = 404
    code = 'not_found'


class ConflictError(AppError):
    status_code = 409
    code = 'conflict'


app = FastAPI()


@app.exception_handler(AppError)
def handle_app_error(request: Request, error: AppError) -> JSONResponse:
    return JSONResponse(
        status_code=error.status_code, content={'detail': error.message, 'code': error.code}
    )


# --- business code: no HTTP anywhere ---
ENROLLED = {(1, 2)}


def enroll(student_id: int, course_id: int) -> None:
    if course_id > 10:
        raise NotFoundError(f'Course {course_id} not found')
    if (student_id, course_id) in ENROLLED:
        raise ConflictError('Already enrolled')
    ENROLLED.add((student_id, course_id))


@app.post('/courses/{course_id}/students/{student_id}', status_code=201)
def enroll_endpoint(course_id: int, student_id: int) -> dict[str, int]:
    enroll(student_id, course_id)
    return {'student_id': student_id, 'course_id': course_id}

یک handler برای همه خطاهای کسب‌وکار: نمونه 06_exception_handlers/app.py

حالا هر کدی، در هر عمقی، می‌تواند raise NotFoundError(...) بزند و کلاینت همیشه یک شکل JSON می‌گیرد: {"detail": "...", "code": "not_found"}.

۷. تنظیمات با pydantic-settings

کلید مخفی، نشانی پایگاه داده و فهرست مبداهای مجاز نباید در کد نوشته شوند. آنها از محیط یا از فایل .env می‌آیند که هرگز در Git قرار نمی‌گیرد.

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file='.env', env_file_encoding='utf-8')

    app_name: str = 'Training Center API'
    debug: bool = False
    admin_email: str = 'admin@kadoos.ir'
    max_page_size: int = Field(default=50, gt=0, le=500)
    secret_key: str = Field(min_length=16)  # required: no default


@lru_cache
def get_settings() -> Settings:
    return Settings()


SettingsDep = Annotated[Settings, Depends(get_settings)]
app = FastAPI()


@app.get('/info')
def info(settings: SettingsDep) -> dict:
    # Never return the secret itself!
    return {
        'app_name': settings.app_name,
        'debug': settings.debug,
        'max_page_size': settings.max_page_size,
        'secret_key_length': len(settings.secret_key),
    }

تنظیمات تایپ‌دار: نمونه 07_settings/app.py

Settings() مقدارها را از متغیرهای محیطی می‌خواند، به نوع درست تبدیل می‌کند و اگر مقدار اجباری نباشد، هنگام راه‌اندازی با صدای بلند شکست می‌خورد، نه وسط یک درخواست. @lru_cache شیء را یک بار می‌سازد و چون get_settings یک وابستگی است، در آزمون‌های فصل سیزدهم می‌توان آن را جایگزین کرد.

وابستگی‌ها را می‌توان برای کل یک روتر یا کل برنامه هم تعریف کرد: APIRouter(dependencies=[Depends(require_api_key)]). با این کار همه endpointهای آن روتر محافظت می‌شوند و فراموش کردن محافظت روی endpoint تازه ممکن نیست (نمونه ۰۸).

۸. ساختار پروژه: هر فایل، یک سؤال

09_project/
├── main.py              سرور را راه می‌اندازد
├── .env.example         به .env کپی کنید
└── app/
    ├── main.py          برنامه FastAPI را می‌سازد
    ├── config.py        تنظیمات
    ├── exceptions.py    خطاهای کسب‌وکار و handler آنها
    ├── schemas.py       مدل‌های Pydantic (از فصل ۳)
    ├── store.py         داده در حافظه (پایگاه داده در فصل ۵)
    ├── services.py      قواعد کسب‌وکار، بدون HTTP
    ├── deps.py          وابستگی‌ها: تنظیمات، سرویس‌ها، صفحه‌بندی، کلید مدیر
    └── routers/         لایه نازک HTTP: درخواست را بخوان، سرویس را صدا بزن

این تنها ساختار درست دنیا نیست. اما هر فایل به یک سؤال پاسخ می‌دهد و می‌دانید کجا را نگاه کنید. همه این فایل‌ها با همین وظیفه تا پایان دوره باقی می‌مانند.

SettingsDep = Annotated[Settings, Depends(get_settings)]
StoreDep = Annotated[Store, Depends(get_store)]


@dataclass
class Paging:
    offset: int
    limit: int


def get_paging(settings: SettingsDep, page: int = 1, size: int = 10) -> Paging:
    size = max(1, min(size, settings.max_page_size))
    return Paging(offset=(max(page, 1) - 1) * size, limit=size)


PagingDep = Annotated[Paging, Depends(get_paging)]


def get_course_service(store: StoreDep) -> CourseService:
    return CourseService(store)


def get_user_service(store: StoreDep) -> UserService:
    return UserService(store)


CourseServiceDep = Annotated[CourseService, Depends(get_course_service)]
UserServiceDep = Annotated[UserService, Depends(get_user_service)]


def get_enrollment_service(
    store: StoreDep, courses: CourseServiceDep, users: UserServiceDep
) -> EnrollmentService:
    return EnrollmentService(store, courses, users)


EnrollmentServiceDep = Annotated[EnrollmentService, Depends(get_enrollment_service)]


def require_admin_key(
    settings: SettingsDep, x_api_key: Annotated[str | None, Header()] = None
) -> None:
    """Temporary "authentication": sessions 9-10 replace it with real users + JWT."""
    if x_api_key != settings.admin_api_key:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail='Invalid or missing X-API-Key')


AdminOnly = Depends(require_admin_key)

همه وابستگی‌ها در یک فایل: 09_project/app/deps.py

@router.get('')
def list_courses(
    service: CourseServiceDep, paging: PagingDep, level: Level | None = None, q: str | None = None
) -> list[CourseRead]:
    return service.list_courses(level=level, q=q, offset=paging.offset, limit=paging.limit)


@router.get('/{course_id}')
def get_course(course_id: int, service: CourseServiceDep) -> CourseRead:
    return service.get(course_id)


@router.post('', status_code=status.HTTP_201_CREATED, dependencies=[AdminOnly])
def create_course(data: CourseCreate, service: CourseServiceDep) -> CourseRead:
    return service.create(data)


@router.patch('/{course_id}', dependencies=[AdminOnly])
def update_course(course_id: int, data: CourseUpdate, service: CourseServiceDep) -> CourseRead:
    return service.update(course_id, data)


@router.delete('/{course_id}', status_code=status.HTTP_204_NO_CONTENT, dependencies=[AdminOnly])
def delete_course(course_id: int, service: CourseServiceDep) -> None:
    service.delete(course_id)

روتر نازک: 09_project/app/routers/courses.py

در روترها هیچ ifی نیست. همه تصمیم‌ها در سرویس‌هاست:

    def enroll(self, course_id: int, student_id: int) -> EnrollmentRead:
        course = self.courses.get(course_id)
        student = self.users.get(student_id)
        if student.role != 'student':
            raise BusinessRuleError('Only students can enroll')
        active = self.for_course(course_id)
        if any(e.student_id == student_id for e in active):
            raise ConflictError('Already enrolled')
        if len(active) >= course.capacity:
            raise ConflictError('Course is full')
        enrollment = EnrollmentRead(
            id=self.store.next_id(self.store.enrollments),
            student_id=student_id,
            course_id=course_id,
            created_at=now(),
        )
        self.store.enrollments[enrollment.id] = enrollment
        return enrollment

قاعده ثبت‌نام در سرویس: 09_project/app/services.py

۹. خلاصه فصل