مرور پایتون پیشرفته، HTTP و REST

۱. چرا مرور، و چرا این موضوع‌ها

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

در کلاس هر مثال را اجرا می‌کنیم، تغییرش می‌دهیم و دوباره اجرا می‌کنیم. توضیح هر بخش کوتاه است؛ کار اصلی با شماست.

۲. راهنمای نوع: زبانی که FastAPI می‌خواند

در دوره پیشرفته، راهنمای نوع بیشتر برای مستندسازی بود. در FastAPI راهنمای نوع دستور است. وقتی می‌نویسید course_id: int، FastAPI مقدار رسیده از آدرس را به عدد تبدیل می‌کند و اگر نشد، پیش از اجرای تابع شما خطای ۴۲۲ برمی‌گرداند. پس از امروز: هر پارامتر و هر مقدار بازگشتی نوع دارد.

from typing import Literal

# A type alias: give a complicated type a name once, and reuse it.
Price = int  # Toman. Money is never a float!
Level = Literal['beginner', 'intermediate', 'advanced']  # only these 3 strings


def average_price(prices: list[Price]) -> float:
    """`list[Price]` in, `float` out. An empty list is handled, not crashed."""
    if not prices:
        return 0.0
    return sum(prices) / len(prices)


def find_course(courses: dict[int, str], course_id: int) -> str | None:
    """`str | None` tells every caller: "be ready, I may have nothing for you"."""
    return courses.get(course_id)


def describe(title: str, level: Level = 'beginner', tags: list[str] | None = None) -> str:
    """Never use a mutable default like `tags: list[str] = []` (it is shared!)."""
    tags = tags or []
    tag_text = ', '.join(tags) if tags else 'no tags'
    return f'{title} [{level}] ({tag_text})'


def split_name(full_name: str) -> tuple[str, str]:
    """A tuple with a fixed shape: (first name, last name)."""
    first, _, last = full_name.strip().partition(' ')
    return first, last

املای امروزی نوع‌ها: نمونه 01_type_hints/app.py

• به جای List[str] و Optional[str] از list[str] و str | None استفاده می‌کنیم.
• Literal فقط چند مقدار مشخص را می‌پذیرد. FastAPI همین را به یک فهرست انتخابی در /docs تبدیل می‌کند.
• نوع str | None به هر فراخواننده می‌گوید: «آماده باش، ممکن است چیزی نداشته باشم». ویرایشگر مجبورتان می‌کند حالت None را بررسی کنید.

۳. Annotated: نوع به‌علاوه قاعده

Annotated[int, ...] از نظر پایتون همان int است؛ بخش دوم را پایتون نادیده می‌گیرد، اما ابزارهای دیگر می‌توانند آن را بخوانند. FastAPI دقیقاً همین کار را می‌کند:

def list_courses(page: Annotated[int, Query(ge=1)] = 1): ...
def get_me(user: Annotated[User, Depends(get_current_user)]): ...

در مثال این بخش، نسخه کوچکی از همین ایده را خودمان می‌سازیم: قاعده را به نوع می‌چسبانیم و بعد با get_type_hints(..., include_extras=True) آن را می‌خوانیم و آرگومان‌ها را بررسی می‌کنیم.

@dataclass(frozen=True)
class Between:
    """A rule object that we will attach to a type."""

    low: int
    high: int


@dataclass(frozen=True)
class MinLength:
    size: int


# Reusable annotated types: define the rule ONCE, use it everywhere.
Capacity = Annotated[int, Between(1, 500)]
Title = Annotated[str, MinLength(3)]


def validate_call(func: Callable[..., object], **arguments: object) -> list[str]:
    """Check `arguments` against the rules found in the hints of `func`."""
    errors: list[str] = []
    hints = get_type_hints(func, include_extras=True)
    for name, value in arguments.items():
        hint = hints.get(name)
        if get_origin(hint) is not Annotated:
            continue
        base_type, *rules = get_args(hint)
        if not isinstance(value, base_type):
            errors.append(f'{name}: expected {base_type.__name__}, got {type(value).__name__}')
            continue
        for rule in rules:
            if isinstance(rule, Between) and not rule.low <= value <= rule.high:
                errors.append(f'{name}: must be between {rule.low} and {rule.high}')
            if isinstance(rule, MinLength) and len(value) < rule.size:
                errors.append(f'{name}: must have at least {rule.size} characters')
    return errors


def create_course(title: Title, capacity: Capacity) -> str:
    return f'created {title!r} with {capacity} seats'

FastAPI در چهل خط: نمونه 02_annotated/app.py

OK       created 'FastAPI' with 20 seats
REJECTED {'title': 'Py', 'capacity': 0} -> ['title: must have at least 3 characters', 'capacity: must be between 1 and 500']
REJECTED {'title': 'SQL', 'capacity': '20'} -> ['capacity: expected int, got str']

۴. جنریک‌ها: یک کلاس برای همه نوع‌ها

کاربر، دوره و ثبت‌نام را باید جایی نگه داریم. نوشتن سه کلاس تقریباً یکسان کار درستی نیست. کلاس جنریک نوع را به عنوان پارامتر می‌گیرد. نحو پایتون ۳٫۱۲ class Repository[T: HasId] است: T هر نوعی می‌تواند باشد که صفت id دارد. «داشتن id» را با یک Protocol بیان می‌کنیم: لازم نیست کلاس از چیزی ارث ببرد، کافی است آن صفت را داشته باشد.

class HasId(Protocol):
    id: int


class Repository[T: HasId]:
    """An in-memory store. Later in the course a database replaces the dict."""

    def __init__(self) -> None:
        self._items: dict[int, T] = {}

    def add(self, item: T) -> T:
        self._items[item.id] = item
        return item

    def get(self, item_id: int) -> T | None:
        return self._items.get(item_id)

    def filter(self, predicate: Callable[[T], bool]) -> list[T]:
        return [item for item in self._items.values() if predicate(item)]

    def __len__(self) -> int:
        return len(self._items)

مخزن جنریک: نمونه 03_generics/app.py

Repository[User]() و Repository[Course]() دو شیء از یک کلاس‌اند، اما ویرایشگر می‌داند users.get(1) یک User برمی‌گرداند. در فصل پنجم همین ایده را در Mapped[int] و در فصل هشتم در Page[CourseRead] می‌بینیم.

۵. dataclass و StrEnum: مدل‌های داده تایپ‌دار

بک‌اند بیشتر وقتش را صرف جابه‌جا کردن داده می‌کند: کاربر، دوره، ثبت‌نام. @dataclass از روی فیلدهای نوع‌دار، سازنده و __repr__ و __eq__ را می‌نویسد.

class Role(StrEnum):
    """A StrEnum is a str too: Role.ADMIN == "admin" is True. Great for JSON."""

    ADMIN = 'admin'
    INSTRUCTOR = 'instructor'
    STUDENT = 'student'


def utc_now() -> datetime:
    return datetime.now(UTC)


@dataclass(slots=True, kw_only=True)
class User:
    id: int
    email: str
    full_name: str
    role: Role = Role.STUDENT
    is_active: bool = True
    created_at: datetime = field(default_factory=utc_now)

    def __post_init__(self) -> None:
        self.email = self.email.strip().lower()
        if '@' not in self.email:
            raise ValueError(f'invalid email: {self.email!r}')


@dataclass(frozen=True, slots=True)
class Money:
    amount: int
    currency: str = 'IRT'

    def __add__(self, other: 'Money') -> 'Money':
        if other.currency != self.currency:
            raise ValueError('cannot add different currencies')
        return Money(self.amount + other.amount, self.currency)


@dataclass(slots=True, kw_only=True)
class Course:
    id: int
    title: str
    price: Money
    capacity: int
    student_ids: list[int] = field(default_factory=list)  # NOT `= []`

    @property
    def seats_left(self) -> int:
        return self.capacity - len(self.student_ids)

مدل‌ها با dataclass: نمونه 04_dataclasses/app.py

• StrEnum هم enum است و هم رشته: Role.ADMIN == 'admin' درست است. برای JSON عالی است.
• slots=True حافظه کمتری می‌گیرد و جلوی اشتباه تایپی مثل user.emial = ... را می‌گیرد.
• frozen=True یک «شیء مقدار» تغییرناپذیر می‌سازد، مثل مبلغ پول.
• برای فهرست پیش‌فرض همیشه field(default_factory=list)؛ هرگز = [].
• در __post_init__ داده را تمیز یا رد می‌کنیم.

۶. توابع مرتبه بالا و بستار

تابع مرتبه بالا تابعی است که تابع می‌گیرد یا تابع برمی‌گرداند. FastAPI روی همین ایده ساخته شده است: شما تابعتان را به FastAPI می‌دهید و FastAPI هر وقت لازم شد صدایش می‌زند. Depends(get_db) هم یعنی «این تابع را صدا بزن و نتیجه‌اش را به من بده».

def make_max_price_filter(max_price: int) -> CoursePredicate:
    """Returns a NEW function that remembers `max_price` (a closure)."""

    def predicate(course: Course) -> bool:
        return course.price <= max_price

    return predicate


def make_level_filter(level: str) -> CoursePredicate:
    return lambda course: course.level == level


def all_of(*predicates: CoursePredicate) -> CoursePredicate:
    """Combine many filters into one -- this is how search filters are built."""
    return lambda course: all(predicate(course) for predicate in predicates)


def apply_discount(percent: int, price: int) -> int:
    return price * (100 - percent) // 100

ساختن فیلتر با بستار: نمونه 05_higher_order_functions/app.py

make_max_price_filter(2_600_000) یک تابع تازه برمی‌گرداند که max_price را به خاطر دارد؛ این یک بستار (closure) است. all_of چند فیلتر را به یکی تبدیل می‌کند: همان کاری که بعدها برای جست‌وجوی دوره‌ها با شرط‌های اختیاری انجام می‌دهیم. در فصل دهم، require_roles('admin') دقیقاً همین شکل را دارد.

۷. دکوراتور، درست نوشته‌شده

@log_calls بالای یک تابع یعنی enroll = log_calls(enroll). سه چیز در هر دکوراتور این دوره رعایت می‌شود:

1. @functools.wraps(func) نام و مستند تابع اصلی را نگه می‌دارد. FastAPI امضای توابع شما را می‌خواند؛ دکوراتوری که امضا را پنهان کند، FastAPI را خراب می‌کند.
2. ParamSpec با نحو [**P, R] نوع پارامترها را نگه می‌دارد، پس ویرایشگر هنوز می‌داند enroll دو عدد صحیح می‌گیرد.
3. استثناها دوباره پرتاب می‌شوند و هرگز بلعیده نمی‌شوند.

def log_calls[**P, R](func: Callable[P, R]) -> Callable[P, R]:
    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        logger.info('-> %s%s', func.__name__, args)
        try:
            result = func(*args, **kwargs)
        except Exception as error:
            logger.warning('<- %s raised %r', func.__name__, error)
            raise
        logger.info('<- %s returned %r', func.__name__, result)
        return result

    return wrapper

دکوراتور ثبت فراخوانی‌ها: نمونه 06_decorators/app.py

کارخانه دکوراتور آرگومان می‌گیرد و خودش دکوراتور برمی‌گرداند: @retry(times=3).

def retry[**P, R](times: int) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """A decorator factory: `@retry(3)` returns the real decorator."""

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except ConnectionError:
                    if attempt == times:
                        raise
                    logger.info('attempt %d failed, retrying', attempt)
            raise AssertionError('unreachable')

        return wrapper

    return decorator

کارخانه دکوراتور: نمونه 06_decorators/app.py

۸. استثناهای سفارشی: خطاها به زبان کسب‌وکار

یک سرویس هرگز با برگرداندن None یا -1 یا رشته "error" نمی‌گوید که کار خراب شد. استثنایی پرتاب می‌کند که دقیقاً می‌گوید چه چیزی خراب شد. سلسله‌مراتب زیر در فصل چهارم به کدهای وضعیت HTTP وصل می‌شود.

class TrainingCenterError(Exception):
    """Base class of all our expected errors."""

    status_code = 400


class NotFoundError(TrainingCenterError):
    status_code = 404

    def __init__(self, entity: str, entity_id: int) -> None:
        super().__init__(f'{entity} {entity_id} was not found')
        self.entity = entity
        self.entity_id = entity_id


class ConflictError(TrainingCenterError):
    status_code = 409


class PermissionDeniedError(TrainingCenterError):
    status_code = 403

سلسله‌مراتب خطاها: نمونه 07_custom_exceptions/app.py

def get_course(course_id: int) -> Course:
    try:
        return COURSES[course_id]
    except KeyError:
        # The KeyError is an implementation detail: hide it.
        raise NotFoundError('Course', course_id) from None


def enroll(student_id: int, course_id: int) -> None:
    course = get_course(course_id)
    if student_id in course.students:
        raise ConflictError(f'student {student_id} is already enrolled')
    if len(course.students) >= course.capacity:
        raise ConflictError(f'course {course.title!r} is full')
    course.students.add(student_id)


def load_price(text: str) -> int:
    try:
        return int(text)
    except ValueError as error:
        # Keep the cause: the traceback will show both errors.
        raise TrainingCenterError(f'invalid price {text!r}') from error

پرتاب خطای مناسب: نمونه 07_custom_exceptions/app.py

• raise ... from None علت درونی بی‌ربط (اینجا KeyError) را پنهان می‌کند.
• raise ... from error علت اصلی را در ردیابی نگه می‌دارد.
• هیچ‌جای این کد از HTTP خبر ندارد. این جداسازی دغدغه‌ها است و در فصل چهارم ارزشش را می‌بینیم.

۹. مدیر زمینه: همیشه پاک‌سازی کن

الگوی «آماده کن، کار را انجام بده، همیشه پاک‌سازی کن» همه‌جای بک‌اند دیده می‌شود: with Session(engine) در فصل پنجم، async with httpx.AsyncClient() در فصل پانزدهم، وابستگی‌های yield در فصل چهارم و lifespan در فصل دوازدهم. در این مثال یک «تراکنش» کوچک می‌سازیم: تغییرات فقط وقتی می‌مانند که بلوک بدون خطا تمام شود.

@contextmanager
def transaction(db: dict[str, list[str]]) -> Iterator[dict[str, list[str]]]:
    """Generator-based context manager: everything before `yield` is __enter__."""
    snapshot = copy.deepcopy(db)
    print('BEGIN')
    try:
        yield db
    except Exception:
        db.clear()
        db.update(snapshot)
        print('ROLLBACK')
        raise
    else:
        print('COMMIT')

تراکنش با @contextmanager: نمونه 08_context_managers/app.py

BEGIN
COMMIT
{'students': ['sara', 'reza']}
BEGIN
ROLLBACK
error: payment failed half way
{'students': ['sara', 'reza']}

۱۰. مولدها: یکی‌یکی، فقط وقتی لازم است

تابعی که yield دارد، به جای ساختن کل فهرست در حافظه، مقدارها را یکی‌یکی تحویل می‌دهد. در این مثال صفحه‌بندی یک منبع داده را از چشم فراخواننده پنهان می‌کنیم. در فصل چهاردهم کلاینت ما دقیقاً همین کار را با صفحه‌های API انجام می‌دهد.

def fetch_page(page: int, size: int) -> list[int]:
    """Pretend to be an API: returns one page of ids (empty at the end)."""
    start = (page - 1) * size
    print(f'  ...fetching page {page}')
    return ALL_COURSE_IDS[start : start + size]


def iterate_all(size: int = 10) -> Iterator[int]:
    """Hide pagination from the caller: they just see a stream of ids."""
    page = 1
    while items := fetch_page(page, size):
        yield from items
        page += 1

پنهان کردن صفحه‌بندی: نمونه 09_generators/app.py

۱۱. async و await: منتظر ماندن هم‌زمان

یک بک‌اند بیشتر وقتش منتظر است: منتظر پایگاه داده، منتظر یک API دیگر، منتظر یک فایل. asyncio.gather چند انتظار را هم‌زمان پیش می‌برد، به شرط اینکه هیچ‌کدام مسدودکننده نباشد.

async def fetch_seats(course_id: int) -> int:
    """Pretend network call that takes 0.5 s."""
    await asyncio.sleep(0.5)
    return 10 + course_id


async def sequential(ids: list[int]) -> list[int]:
    return [await fetch_seats(course_id) for course_id in ids]


async def concurrent(ids: list[int]) -> list[int]:
    return list(await asyncio.gather(*(fetch_seats(course_id) for course_id in ids)))


async def blocking_mistake(ids: list[int]) -> None:
    """time.sleep() inside async code: gather cannot help, the loop is frozen."""

    async def bad(course_id: int) -> None:
        time.sleep(0.5)  # WRONG inside async def

    await asyncio.gather(*(bad(course_id) for course_id in ids))

ترتیبی، هم‌زمان و اشتباه رایج: نمونه 10_async_await/app.py

sequential  2.00s  [11, 12, 13, 14]
concurrent  0.50s  [11, 12, 13, 14]
blocking    2.00s  (gather did not help!)
gave up after 0.2 s

۱۲. HTTP، REST و JSON

در مدل کلاینت و سرور، کلاینت یک درخواست می‌فرستد و سرور یک پاسخ برمی‌گرداند. هر درخواست یک روش، یک نشانی، چند هدر و گاهی یک بدنه دارد. هر پاسخ یک کد وضعیت، چند هدر و یک بدنه دارد.

REST یعنی منابع (اسم‌ها) در نشانی و عمل در روش: GET /courses فهرست، POST /courses ساختن، GET /courses/1 یکی، PUT جایگزین کردن کامل، PATCH تغییر بخشی و DELETE حذف. JSON قالب متنی بدنه است.

def main() -> None:
    # JSON is just text:
    text = json.dumps({'title': 'FastAPI', 'price': 4_800_000, 'tags': ['web']})
    print(text, '->', json.loads(text)['price'])

    # One Client = one connection pool, reused for every request.
    with httpx.Client(base_url=BASE_URL, timeout=5.0) as client:
        response = client.get('/posts', params={'userId': 1})
        print(response.request.method, response.request.url)
        print(response.status_code, response.headers['content-type'])
        posts = [Post.from_json(item) for item in response.json()]
        print(f'{len(posts)} posts, first: {posts[0].title!r}')

        created = client.post('/posts', json={'title': 'Hello', 'body': '...', 'userId': 1})
        print('POST ->', created.status_code, created.json())

        missing = client.get('/posts/999999')
        print('GET missing ->', missing.status_code)

        try:
            missing.raise_for_status()
        except httpx.HTTPStatusError as error:
            print('raise_for_status():', error.response.status_code)

    try:
        httpx.get(f'{BASE_URL}/posts', timeout=0.001)
    except httpx.TimeoutException:
        print('timeout: never wait for ever')

اولین کلاینت با HTTPX: نمونه 11_http_and_json/app.py

۱۳. Bruno: اولین درخواست‌ها

پوشه 13_bruno_first_requests یک مجموعه Bruno است. آن را با Open Collection باز کنید، محیط public را انتخاب کنید و هر چهار درخواست را اجرا کنید: یک GET با پارامتر پرس‌وجو، یک GET تکی، یک POST با بدنه JSON و یک GET که ۴۰۴ می‌گیرد.

۱۴. بازنویسی یک کد بد

فایل 12_refactor_me/app.py کار می‌کند و گزارش درستی چاپ می‌کند. همین خطرناک است: «کار می‌کند» با «خوب است» فرق دارد. نبود راهنمای نوع، دیکشنری‌های جادویی با کلیدهای 't' و 'p'، متغیر سراسری، except: خالی، آرگومان پیش‌فرض تغییرپذیر و پول اعشاری، همه در آن هست. تمرین ۱۲ از شما می‌خواهد آن را بدون تغییر خروجی بازنویسی کنید.

۱۵. پروژه: هسته آموزشگاه با پایتون خالص

پروژه‌ای که شانزده جلسه با ما همراه است امروز بدون هیچ چارچوبی شروع می‌شود: مدل‌ها با dataclass، یک مخزن جنریک، یک سرویس با قواعد کسب‌وکار و استثناهای سفارشی.

    def enroll(self, student_id: int, course_id: int) -> Course:
        student = self.users.get(student_id)
        course = self.courses.get(course_id)
        if student.role is not Role.STUDENT or not student.is_active:
            raise PermissionDeniedError('only active students can enroll')
        if student.id in course.student_ids:
            raise AlreadyEnrolledError(f'user {student.id} is already in course {course.id}')
        if course.is_full:
            raise CourseFullError(f'course {course.id} is full')
        course.student_ids.add(student.id)
        return course

قاعده ثبت‌نام: 14_project/training_center/services.py

سرویس از ترکیب (composition) استفاده می‌کند: دو مخزن را در خودش دارد و از آنها ارث نمی‌برد. در فصل دوم همین قواعد پشت endpointها می‌روند و در فصل پنجم دیکشنری درون مخزن جایش را به پایگاه داده می‌دهد؛ اما قواعد و استثناها همین می‌مانند.

refused -> CourseFullError: course 2 is full
refused -> AlreadyEnrolledError: user 2 is already in course 1
refused -> NotFoundError: Course 99 not found
refused -> PermissionDeniedError: only instructors can create courses
rolled back: still 3 users

۱۶. خلاصه فصل