Нэгдсэн төсөл: Салбаруудын захиалга цуглуулагч

Танилцуулга

Энэ бол сургалтын сүүлчийн хичээл бөгөөд таван түвшний бүх хэрэгслийг нэг програмд нэгтгэх төсөл. Өмнөх build-үүд тус бүр нэг түвшинд төвлөрсөн: log боловсруулагч iteration-д, хэмжигч хэрэгслийн сан function-д, мэдэгдлийн систем OOP-д, импортлогч төрөл, алдаа, хурданд. Энэ төсөл тэдгээрийг бүгдийг шаардана, учир нь бодит програм нэг сэдвээр хязгаарлагддаггүй.

Төсөл нь салбаруудын захиалга цуглуулагч. Дэлгүүрийн сүлжээ зургаан аймаг, хотод салбартай. Салбар бүр өөрийн сервертэй бөгөөд захиалгуудаа хуудаслан, нэг хуудсанд тавин захиалгаар, HTTP-ээр өгдөг. Төв оффист өдөр бүр бүх салбарын захиалгыг цуглуулж, хураангуй тайлан гаргах хэрэгтэй. Бодит нөхцөлийн бүх бэрхшээл бий: нэг салбарын сервер үе үе "түр боломжгүй" гэж хариулдаг, нөгөөгийнх хэт удаан, заримдаа огт хариулдаггүй. Цуглуулагч эдгээрийг даван, авч чадсан бүх зүйлийг авч, юу дутуу байгааг шударгаар тайлагнах ёстой.

Бид төслийг хоёр хэсгээр бүтээнэ. Эхлээд MockTransport-оор хуурамч салбарын серверүүд: Хичээл 15.2-ын хэрэгсэл, энд санаатайгаар тогтвортой бус. Дараа нь цуглуулагч өөрөө: Түвшин 3-ын dataclass ба classmethod, Түвшин 4-ийн typing ба exception шатлал, Түвшин 2-ын argument-тай async decorator ба context manager, Түвшин 1-ийн хуудаслалтыг нуудаг async generator, Түвшин 5-ын TaskGroup, Semaphore, timeout. Төгсгөлд нь хэмжилт, mypy-ийн strict горим, pytest тестүүд, commit, мөн бүх сургалтын тойм.

Зүйрлэл

Аймгийн төвүүдээс тайлан цуглуулж буй яамны хэлтсийг төсөөлөөд үзээрэй. Хэлтэс зургаан аймагт нэг дор хүсэлт илгээнэ: нэг нэгээр нь хариу хүлээвэл долоо хоног зарцуулна. Гэхдээ бүх аймагт нэг дор шахалт үзүүлвэл утасны шугам ачаалагдана: нэг удаад хэдийг дуудахыг хязгаарлана. Аймаг бүрийн тайлан хэдэн хуудастай, хуудас бүрийг тусад нь илгээнэ.

Нэг аймгийн шугам тасарвал хэлтэс дахин залгана, гэхдээ хязгааргүй биш: гурван оролдлогын дараа бууж өгнө. Нэг аймаг хариулахгүй бол хэлтэс бусдыгаа хүлээлгэхгүй: тэр аймгаас авсан хэсгийг хадгалж, тайланд "дутуу, шалтгаан нь" гэж тэмдэглэнэ. Харин хэрэв хэлтсийн өөрийн маягт буруу хэвлэгдсэн байвал, хэлтсийн дотоод алдаа, ажлыг зогсоож засна: түүнийг "аймгийн асуудал" гэж нууж болохгүй.

Энэ зүйрлэлд төслийн бүх дизайны шийдвэр бий. Зэрэг хүсэлт бол TaskGroup. Шугамын хязгаар бол Semaphore. Хуудас хуудсаар бол async generator. Гурван оролдлого бол retry. Хүлээх хязгаар бол timeout. Дутуу тэмдэглэх бол Хичээл 12.4-ийн "алдаатай мөрийг цуглуулж үргэлжлүүл". Дотоод алдааг нуухгүй байх бол Хичээл 11.1-ийн "зөвхөн юу хийхээ мэддэгээ бар".

Алхам 1: branch ба хуурамч серверүүд

git switch -c level5-collector

Шинэ хавтаст httpx2 суусан virtual environment-ийг идэвхжүүл. fake_branches.py файл:

python

import asyncio
import random

import httpx2

PAGES = 3
PER_PAGE = 50
STATUSES = ["paid", "paid", "pending", "cancelled"]


def make_transport(
    seed: int = 42,
    latency: float = 0.2,
    slow_branch: str = "darkhan",
    flaky_branch: str = "erdenet",
) -> httpx2.MockTransport:
    failures_left = {flaky_branch: 1}

    async def handler(request: httpx2.Request) -> httpx2.Response:
        parts = request.url.path.strip("/").split("/")
        if len(parts) != 3 or parts[0] != "branches" or parts[2] != "orders":
            return httpx2.Response(404)
        branch = parts[1]
        page = int(request.url.params.get("page", "1"))

        await asyncio.sleep(latency)
        if branch == slow_branch and page == 2:
            await asyncio.sleep(10.0)
        if failures_left.get(branch, 0) > 0:
            failures_left[branch] -= 1
            return httpx2.Response(503)

        rng = random.Random(f"{seed}-{branch}-{page}")
        orders = [
            {
                "id": f"{branch}-{page}-{i}",
                "amount": rng.randint(1_000, 200_000),
                "status": rng.choice(STATUSES),
            }
            for i in range(PER_PAGE)
        ]
        next_page = page + 1 if page < PAGES else None
        return httpx2.Response(200, json={"orders": orders, "next_page": next_page})

    return httpx2.MockTransport(handler)

Хичээл 15.2-ын fake_shop-оос гурван ялгаа бий. Нэгдүгээрт, handler async def: MockTransport async функцийг ч хүлээн авдаг бөгөөд asyncio.sleep-ээр сүлжээний саатлыг, latency, загварчилна. Хоёрдугаарт, хоёр "муу" салбар. Дархан хоёр дахь хуудсандаа арван секунд гацна: хариулахгүй сервер. Эрдэнэт анхны хүсэлтэд нэг удаа 503 буцаана: түр саатал. failures_left dict нь Бүлэг 4-ийн closure-оор handler-т хадгалагдсан төлөв. Гуравдугаарт, хуудас бүрийн өгөгдөл өөрийн random.Random-тай бөгөөд seed нь салбар ба хуудаснаас бүрдэнэ. Нэг нийтлэг rng хэрэглэсэн бол зэрэг хүсэлтүүдийн дараалал ажиллуулах бүрт өөр байх тул дүнгүүд ч өөр гарна: Хичээл 13.2-ын "үр дүн ажиллуулах бүрт өөр" асуудлын бас нэг хэлбэр. Тусдаа seed нь хүсэлт ямар дарааллаар ирснээс үл хамааран ижил өгөгдөл баталгаажуулна.

Хуудаслалтын гэрээ: хариу бүр захиалгуудын жагсаалт ба дараагийн хуудасны дугаартай, сүүлийн хуудсанд дугаарын оронд null. Ихэнх бодит API яг ийм хэлбэртэй.

Алхам 2: төрлүүд ба exception шатлал

collector.py файл. Эхлээд импорт, төрлүүд, алдаанууд:

python

import asyncio
import functools
import os
import time
from collections.abc import AsyncIterator, Awaitable, Callable, Iterator
from contextlib import contextmanager
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Literal, Self, TextIO, get_args

import httpx2

Status = Literal["paid", "pending", "cancelled"]
STATUS_BY_NAME: dict[str, Status] = {name: name for name in get_args(Status)}
PAGE_TIMEOUT = 1.0


class CollectError(Exception):
    """Салбараас өгөгдөл цуглуулах үеийн бүх алдааны үндэс."""


class FetchTimeoutError(CollectError):
    """Хуудас заасан хугацаанд ирсэнгүй. Дахин оролдож болно."""


class BadStatusError(CollectError):
    """Сервер алдаатай status code буцаалаа."""

    def __init__(self, url: str, status: int) -> None:
        super().__init__(f"{url}: сервер {status} буцаалаа")
        self.url = url
        self.status = status

    @property
    def retryable(self) -> bool:
        return self.status >= 500


@dataclass(frozen=True)
class Order:
    branch: str
    order_id: str
    amount: int
    status: Status

    @classmethod
    def from_dict(cls, branch: str, data: dict[str, Any]) -> Self:
        status = STATUS_BY_NAME.get(str(data["status"]))
        if status is None:
            raise CollectError(f"{branch}: үл мэдэгдэх төлөв {data['status']!r}")
        return cls(branch, str(data["id"]), int(data["amount"]), status)


@dataclass
class BranchResult:
    branch: str
    orders: list[Order] = field(default_factory=list)
    error: CollectError | None = None

Энэ хэсэгт Түвшин 3 ба 4 нийлж байна. Status ба STATUS_BY_NAME нь Хичээл 12.4-ийн загвар. Exception шатлал Хичээл 11.1-ийн хэлбэрээр, docstring бүр алдааны утгыг хэлдэг: дахин оролдох уу, үгүй юу. BadStatusError-ийн retryable нь Хичээл 7.4-ийн property: 500-аас дээш, серверийн алдаа, бол дахин оролдох утгатай, 4xx бол утгагүй. Энэ шийдвэр нэг газар амьдарна.

Order.from_dict бол Хичээл 7.1-ийн classmethod: dict-ээс объект үүсгэх өөр зам. Буцаах төрөл Self нь "энэ класс" гэсэн утгатай, Python 3.11-ээс хойш typing модульд бий: Order-ийн хүүхэд класс from_dict дуудвал mypy буцаах төрлийг хүүхэд класс гэж зөв ойлгоно, Хичээл 7.1-ийн cls-ийн санааны төрлийн хувилбар. BranchResult нь салбар бүрийн үр дүн: авч чадсан захиалгууд ба, хэрэв байвал, юунаас болж зогссон. Алдаа ба хагас өгөгдөл хамт хадгалагдана: зүйрлэлийн "дутуу, шалтгаан нь".

Алхам 3: async retry decorator

Хичээл 5.3-т бид argument-тай retry decorator бичсэн. Энэ удаа async хувилбар, mypy-ийн strict горимд нийцэх төрөлтэй:

python

def retry[**P, T](
    times: int = 3, delay: float = 0.05
) -> Callable[[Callable[P, Awaitable[T]]], Callable[P, Awaitable[T]]]:
    def decorate(func: Callable[P, Awaitable[T]]) -> Callable[P, Awaitable[T]]:
        @functools.wraps(func)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
            attempt = 1
            while True:
                try:
                    return await func(*args, **kwargs)
                except BadStatusError as error:
                    if not error.retryable or attempt >= times:
                        raise
                except FetchTimeoutError:
                    if attempt >= times:
                        raise
                await asyncio.sleep(delay * attempt)
                attempt += 1

        return wrapper

    return decorate

Бүтэц Хичээл 5.3-тай яг ижил гурван давхарга: тохиргооны цаг (retry), боолтын цаг (decorate), дуудлагын цаг (wrapper). Хичээл 5.2-ын functools.wraps нэр ба docstring-ийг хадгална. Хоёр ялгаа бий. wrapper бол async def, функцийг await хийнэ, хүлээлт time.sleep биш asyncio.sleep: Хичээл 14.1-ийн дүрэм. Хүлээлт оролдлого бүрт уртасна, delay × attempt: серверт сэргэх хугацаа өгнө.

Төрлийн тэмдэглэгээнд нэг шинэ зүйл бий: **P. Хичээл 10.2-т T нь нэг утгын төрлийг хадгалдаг төрлийн хувьсагч байсан. P нь ParamSpec: функцийн бүх аргументын хэлбэрийг, хэдэн аргумент, ямар төрлүүд, хадгалдаг төрлийн хувьсагч. P.args ба P.kwargs нь Хичээл 5.1-ийн args ба *kwargs-ийн төрлүүд. Ингэснээр mypy decorator-оор боосон функцийг анхных шигээ шалгана: буруу аргумент дамжуулбал алдааг илрүүлнэ. Энэ синтакс Python 3.12-оос хойш бий.

Retry-ийн бодлогыг уншъя: 404 зэрэг клиентийн алдаанд шууд бууж өгнө, серверийн алдаа ба timeout-д гурван удаа оролдоно. while True хэлбэр нь for давталтаас илүү бөгөөд mypy-д "функц үргэлж return эсвэл raise хийнэ" гэдгийг ойлгомжтой болгодог.

Алхам 4: хуудас татах ба хуудаслалтыг нуух

python

@retry(times=3, delay=0.05)
async def fetch_page(
    client: httpx2.AsyncClient, branch: str, page: int, timeout: float = PAGE_TIMEOUT
) -> dict[str, Any]:
    url = f"/branches/{branch}/orders"
    try:
        async with asyncio.timeout(timeout):
            response = await client.get(url, params={"page": page})
        response.raise_for_status()
    except (TimeoutError, httpx2.TimeoutException) as error:
        raise FetchTimeoutError(f"{url}?page={page}: хугацаа хэтэрлээ") from error
    except httpx2.HTTPStatusError as error:
        raise BadStatusError(url, error.response.status_code) from error
    except httpx2.RequestError as error:
        raise CollectError(f"{url}: холболтын алдаа") from error
    else:
        data: dict[str, Any] = response.json()
        return data


async def iter_orders(
    client: httpx2.AsyncClient, branch: str, timeout: float = PAGE_TIMEOUT
) -> AsyncIterator[Order]:
    page: int | None = 1
    while page is not None:
        data = await fetch_page(client, branch, page, timeout=timeout)
        for raw in data["orders"]:
            yield Order.from_dict(branch, raw)
        page = data["next_page"]

fetch_page нь Хичээл 15.2-ын fetch_json-ийн бүтэцтэй: сүлжээний гурван ангиллын алдааг өөрийн шатлалд raise from-оор хөрвүүлж, try-г else-ээр нарийн байлгана. Нэг ялгаа бий: timeout-ийг client-ийн тохиргоонд биш, Хичээл 14.2-ын asyncio.timeout-оор. Шалтгаан нь шударга: client-ийн timeout-ийг жинхэнэ сүлжээний давхарга хэрэгжүүлдэг бөгөөд MockTransport түүнийг мэддэггүй. asyncio.timeout аль ч транспортод ажиллана. Тиймээс хоёр төрлийн timeout алдааг хоёуланг нь барьсан: бодит сүлжээнд аль нь эхэлж хүрэхээс хамаарна. @retry fetch_page-ийг боосон тул timeout ба 503 автоматаар дахин оролдогдоно.

iter_orders бол Хичээл 15.1-ийн async generator, хуудаслалтыг бүрэн нууна. Дуудагч хуудас, next_page, хэдэн хүсэлт явсан талаар юу ч мэдэхгүй: зүгээр л async for order in iter_orders(...). Энэ бол Түвшин 1-ийн гол санаа, Хичээл 1.4-ийн Batches ба Хичээл 2.3-ын pipeline, эцсийн хэлбэрээрээ: дотоод бүтцийг нуусан урсгал. Order.from_dict-ийн CollectError ч энэ урсгалаас гарна.

Алхам 5: салбар ба бүх салбар

python

async def collect_branch(
    client: httpx2.AsyncClient,
    branch: str,
    semaphore: asyncio.Semaphore,
    timeout: float = PAGE_TIMEOUT,
) -> BranchResult:
    result = BranchResult(branch)
    async with semaphore:
        try:
            async for order in iter_orders(client, branch, timeout):
                result.orders.append(order)
        except CollectError as error:
            result.error = error
    return result


async def collect_all(
    client: httpx2.AsyncClient,
    branches: list[str],
    limit: int = 4,
    timeout: float = PAGE_TIMEOUT,
) -> list[BranchResult]:
    semaphore = asyncio.Semaphore(limit)
    async with asyncio.TaskGroup() as group:
        tasks = [
            group.create_task(collect_branch(client, branch, semaphore, timeout))
            for branch in branches
        ]
    return [task.result() for task in tasks]

collect_branch-ийн try нь зөвхөн CollectError-ийг барина. Энэ бол төслийн хамгийн чухал шийдвэрүүдийн нэг. Салбарын асуудал, timeout, 503, үл мэдэгдэх төлөв, тэр салбарын үр дүнд тэмдэглэгдэж, бусад салбарт нөлөөлөхгүй. Харин KeyError эсвэл TypeError шиг кодын алдаа барьагдахгүй: тэр collect_all-ийн TaskGroup руу гарч, бусад task-уудыг цуцалж, ExceptionGroup болон програмаас гарна. Хичээл 11.1-ийн Жишээ 4-ийн урхи, NameError нуугдах, энд гарахгүй. Тест хэсэгт үүнийг баталгаажуулна.

Хагас өгөгдөл хадгалагдана: Дархан эхний хуудсаа өгөөд хоёр дахьд гацвал эхний тавин захиалга result.orders-д үлдэнэ, error нь яагаад зогссоныг хэлнэ.

Semaphore нэг удаад хэдэн салбартай зэрэг ажиллахыг хязгаарлана: салбар доторх хуудсууд дараалан явна, учир нь хуудас бүр өмнөхийнхөө next_page-ээс хамаардаг. Semaphore ба TaskGroup collect_all дотор, ажиллаж буй loop-д үүснэ: Хичээл 14.3-ын "өөр loop-той холбогдсон" урхинаас зайлсхийж. task.result() нь Хичээл 14.2-ын загвар: TaskGroup-ийн дараа бүх task дууссан нь баталгаатай.

Алхам 6: тайлан, хэмжилт ба main

python

@contextmanager
def atomic_write(path: Path, encoding: str = "utf-8") -> Iterator[TextIO]:
    temp_path = path.with_name(path.name + ".tmp")
    try:
        with temp_path.open("w", encoding=encoding) as f:
            yield f
    except BaseException:
        temp_path.unlink(missing_ok=True)
        raise
    else:
        os.replace(temp_path, path)


@contextmanager
def timer(label: str) -> Iterator[None]:
    start = time.perf_counter()
    try:
        yield
    finally:
        print(f"{label}: {time.perf_counter() - start:.2f} с")


def write_summary(results: list[BranchResult], path: Path) -> None:
    lines: list[str] = []
    for result in results:
        paid = sum(order.amount for order in result.orders if order.status == "paid")
        state = "бүрэн" if result.error is None else f"ДУТУУ: {result.error}"
        lines.append(
            f"{result.branch:<12} {len(result.orders):>4} захиалга  "
            f"төлөгдсөн {paid:>12,}₮  {state}"
        )
    total = sum(len(result.orders) for result in results)
    incomplete = sum(result.error is not None for result in results)
    lines.append(f"Нийт: {total:,} захиалга, {incomplete} салбар дутуу")
    with atomic_write(path) as f:
        f.write("\n".join(lines) + "\n")


BRANCHES = ["ulaanbaatar", "darkhan", "erdenet", "khovd", "choibalsan", "murun"]


async def run(transport: httpx2.MockTransport, limit: int) -> list[BranchResult]:
    async with httpx2.AsyncClient(transport=transport, base_url="https://shop.test") as client:
        return await collect_all(client, BRANCHES, limit=limit)


def main() -> None:
    from fake_branches import make_transport

    for limit in (1, len(BRANCHES)):
        with timer(f"limit={limit}"):
            results = asyncio.run(run(make_transport(), limit))
    write_summary(results, Path("summary.txt"))
    print(Path("summary.txt").read_text(encoding="utf-8"))


if __name__ == "__main__":
    main()

atomic_write нь Хичээл 11.3 ба 12.4-өөс. timer бол Хичээл 6.3-ын хамгийн богино context manager, Хичээл 6.4-ийн measurekit-ийн эхлэл. write_summary нь Хичээл 12.3-ын join загвараар мөрүүдийг цуглуулна. sum(result.error is not None ...) нь True-г 1, False-ийг 0 гэж тоолно.

main хуурамч серверүүдийг функц дотор import хийдэг: collector.py өөрөө тэднээс хамаарахгүй, зөвхөн туршилтын ажиллуулалт тэднийг хэрэглэнэ. Бодит салбарын серверүүдтэй ажиллахад ганц мөр өөрчлөгдөнө: httpx2.AsyncClient(base_url="https://...", timeout=10.0), transport-гүйгээр. Хичээл 15.2-т хэлсэн Protocol-ийн санаа: цуглуулагч сүлжээг шууд мэддэггүй.

python collector.py

Нэг ажиллуулалт:

limit=1: 6.62 с
limit=6: 3.37 с
ulaanbaatar   150 захиалга  төлөгдсөн    7,412,903₮  бүрэн
darkhan        50 захиалга  төлөгдсөн    2,388,115₮  ДУТУУ: /branches/darkhan/orders?page=2: хугацаа хэтэрлээ
erdenet       150 захиалга  төлөгдсөн    7,659,240₮  бүрэн
khovd         150 захиалга  төлөгдсөн    7,021,577₮  бүрэн
choibalsan    150 захиалга  төлөгдсөн    7,840,612₮  бүрэн
murun         150 захиалга  төлөгдсөн    7,203,384₮  бүрэн
Нийт: 800 захиалга, 1 салбар дутуу

Дүнгүүд жишээ: тогтмол seed-тэй тул таны гаралт ажиллуулах бүрт ижил, гэхдээ миний харуулснаас өөр байж болно.

Тайлан бүх зүйлийг шударгаар хэлж байна. Эрдэнэтийн 503 тайланд огт харагдахгүй: retry түүнийг чимээгүй даван туулж, бүрэн 150 захиалга авсан. Дархан эхний хуудсаа өгсөн, хоёр дахьд гурван удаа timeout болж, 50 захиалгатай "дутуу" тэмдэглэгдсэн. Бусад салбар түүнийг хүлээгээгүй.

Хэмжилт: хамгийн удаан салбар

Хичээл 12.1-ийн загвараар хэмжилтийг уншъя. limit=1 бол дараалсан: хуудас бүр 0.2 секунд, хэвийн салбар бүр гурван хуудас буюу 0.6 секунд. Дархан ганцаараа 3.35 секунд: эхний хуудас 0.2, дараа нь гурван timeout тус бүр 1.0 секунд, хооронд нь 0.05 ба 0.1 секундийн retry хүлээлт. Нийт 6.6 секунд. limit=6 бол бүх салбар зэрэг: нийт хугацаа бол хамгийн удаан салбарын хугацаа, 3.37 секунд.

Хурд хоёр дахин л нэмэгдсэн, Хичээл 14.1-ийн арван дахинаас хол. Хэмжилт хамгийн чухал сургамжийг харуулж байна: зэрэгцээ системд нийт хугацааг хамгийн удаан ажил тодорхойлдог. Дарханы хугацааны бараг бүх нь timeout ба retry. Цаашид хурдасгах хэрэгсэл бол илүү олон task биш, бодлогын шийдвэр: timeout-ийг 0.5 секунд болгох уу? Timeout-д хоёр удаа л оролдох уу? Энэ шийдвэр техникийн асуулт биш: удаан салбарын өгөгдлийг хүлээх үнэ цэнэ хэд вэ гэдэг бизнесийн асуулт. Хэмжилт тэр асуултыг тоогоор тавьж өгнө, хариултыг таны байгууллага гаргана.

Алхам 7: mypy strict

pyproject.toml:

toml

[tool.mypy]
strict = true

mypy collector.py fake_branches.py

Success: no issues found in 2 source files

Хоёр файл бүрэн шалгагдсан: ParamSpec-тэй decorator, Self, AsyncIterator, MockTransport-ийн async handler бүгд. mypy-ийн шалгалт retry-ийн боосон fetch_page-ийн аргументуудыг ч хянадаг: fetch_page(client, branch, "2") гэж бичвэл page int биш гэсэн алдаа гарна, decorator-ын дундуур ч.

Алхам 8: тестүүд

test_collector.py. Async кодыг тестлэхэд pytest-д нэмэлт plugin бий, pytest-asyncio, гэвч бидэнд хэрэггүй: тест бүр asyncio.run-аар өөрийн loop-ийг эхлүүлж болно.

python

import asyncio

import httpx2
import pytest

from collector import (
    BadStatusError,
    BranchResult,
    CollectError,
    FetchTimeoutError,
    collect_all,
)
from fake_branches import make_transport


def collect(transport, branches, **kwargs) -> list[BranchResult]:
    async def go():
        async with httpx2.AsyncClient(transport=transport, base_url="https://shop.test") as client:
            return await collect_all(client, branches, **kwargs)
    return asyncio.run(go())


def test_all_pages_are_collected():
    transport = make_transport(latency=0.01, slow_branch="", flaky_branch="")
    [result] = collect(transport, ["ulaanbaatar"])
    assert len(result.orders) == 150
    assert result.error is None


def test_retry_recovers_from_one_503():
    transport = make_transport(latency=0.01, slow_branch="", flaky_branch="erdenet")
    [result] = collect(transport, ["erdenet"])
    assert len(result.orders) == 150
    assert result.error is None


def test_slow_branch_keeps_partial_data():
    transport = make_transport(latency=0.01, slow_branch="darkhan", flaky_branch="")
    [result] = collect(transport, ["darkhan"], timeout=0.1)
    assert len(result.orders) == 50
    assert isinstance(result.error, FetchTimeoutError)
    assert isinstance(result.error.__cause__, TimeoutError)


def test_client_errors_are_not_retried():
    calls = 0

    def not_found(request):
        nonlocal calls
        calls += 1
        return httpx2.Response(404)

    [result] = collect(httpx2.MockTransport(not_found), ["ulaanbaatar"])
    assert isinstance(result.error, BadStatusError)
    assert result.error.status == 404
    assert calls == 1


def test_unknown_status_is_a_collect_error():
    def weird(request):
        order = {"id": 1, "amount": 500, "status": "lost"}
        return httpx2.Response(200, json={"orders": [order], "next_page": None})

    [result] = collect(httpx2.MockTransport(weird), ["khovd"])
    assert type(result.error) is CollectError
    assert "lost" in str(result.error)


def test_code_bugs_are_not_hidden():
    def broken(request):
        return httpx2.Response(200, json={"items": []})

    with pytest.raises(ExceptionGroup) as info:
        collect(httpx2.MockTransport(broken), ["murun"])
    assert isinstance(info.value.exceptions[0], KeyError)


def test_semaphore_limits_concurrent_branches():
    current = peak = 0

    async def counting(request):
        nonlocal current, peak
        current += 1
        peak = max(peak, current)
        await asyncio.sleep(0.01)
        current -= 1
        return httpx2.Response(200, json={"orders": [], "next_page": None})

    collect(httpx2.MockTransport(counting), [f"b{i}" for i in range(6)], limit=2)
    assert peak == 2

Долоон тест төслийн долоон амлалтыг шалгана. Эхний гурав хуурамч серверийн тохиргоогоор: latency=0.01 тестийг хурдан болгоно, хоосон string slow_branch ба flaky_branch "муу салбар байхгүй" гэсэн утгатай. Дарханы тестэд timeout=0.1 өгсөн тул арван секундийн гацалт 0.1 секундэд таслагдана: тест бүхэлдээ секундээс бага. __cause__ шалгалт raise from гинжийг баталгаажуулна.

Сүүлийн дөрөв Хичээл 15.2-ын адил өөрийн жижиг handler-тэй. not_found дуудлагыг Хичээл 4.2-ын nonlocal-оор тоолж, 404 дахин оролдогдоогүйг баталгаажуулна: retryable property ажиллаж байна. weird нь Order.from_dict-ийн шалгалтыг шалгана. type(...) is CollectError нь хүүхэд класс биш, яг үндсэн класс гэдгийг шалгана.

broken нь төслийн хамгийн чухал тест: “orders” түлхүүргүй хариу iter_orders-д KeyError үүсгэнэ, кодын алдаа. Тэр collect_branch-д барьагдахгүй, TaskGroup-ээс ExceptionGroup болж гарах ёстой. Хэрэв хэн нэгэн ирээдүйд collect_branch-ийн except CollectError-ийг except Exception болговол энэ тест бүтэлгүйтэж, алдаа нуугдаж эхэлснийг шууд мэдэгдэнэ. counting handler нь Хичээл 14.3-ын Tracker-ийн санаа: хэдэн хүсэлт зэрэг ирж байгааг серверийн талаас тоолно.

python -m pytest -q

.......                                                                  [100%]
7 passed in 0.74s

Алхам 9: commit

.gitignore:

summary.txt
__pycache__/
.mypy_cache/
.pytest_cache/

git add fake_branches.py collector.py test_collector.py pyproject.toml .gitignore
git commit -m "Түвшин 5: салбаруудын захиалга цуглуулагч — asyncio, httpx2, бүх түвшин нэгдсэн"

Сургалт бүхэлдээ

Энэ төслийн мөр бүрийг таван түвшинтэй холбож болно. Бүхэл зургийг харах нь зүйтэй, учир нь бие даасан хичээл бүр нэг хэрэгсэл заасан бол энэ төсөл тэдгээр хэрхэн хамтран ажилладгийг харуулсан.

Түвшин 1, iteration: iter_orders хуудаслалтыг нуудаг async generator, Хичээл 1.4-ийн өөрийн iterator ба 2.3-ын pipeline-ийн үргэлжлэл. Түвшин 2, function: retry нь closure дээр бүтсэн, wraps-тай, argument-тай decorator. atomic_write ба timer нь contextmanager. failures_left ба nonlocal тоолуурууд closure-ийн төлөв. Түвшин 3, OOP: frozen dataclass Order, classmethod from_dict, property retryable, default_factory-тэй BranchResult, transport-оор солигддог сүлжээ, Protocol-ийн санаа. Түвшин 4, төрөл, алдаа, хурд: Literal, Self, ParamSpec, mypy strict, CollectError шатлал ба raise from, нарийн try ба else, цуглуулж үргэлжлүүлэх ба кодын алдааг нуухгүй байх ялгаа, хэмжилтээр олсон "хамгийн удаан салбар". Түвшин 5, зэрэгцээ ажиллагаа: AsyncClient, TaskGroup, Semaphore, asyncio.timeout, MockTransport.

Сургалтын туршид нэг сахилга бат тасралтгүй байсан. Хурдны тухай мэдэгдэл бүр хэмжилттэй. Алдаа бүр нэр, мессеж, шалтгаантай. Далд төлөвтэй ухагдахуун бүр trace-ээр. Build бүр тест ба commit-оор. Эдгээр нь Python-ий тусгай хэрэгсэл биш, ямар ч хэлэнд, ямар ч төсөлд хамаарах ажлын хэв маяг. Хэрэгслүүд хувьсана: энэ бүлэгт л бид httpx-ээс httpx2 руу шилжсэн, Python-ий GIL өөрчлөгдөж байна. Харин "хэмж", "алдааг нэрлэ", "далд төлөвийг мөрд", "тестлээд хадгал" гэсэн дадал хувьсахгүй.

Туршаад үзээрэй

Хүсвэл цуглуулагчид Хичээл 9.4-ийн EventBus-ийг нэмж үзээрэй: салбар бүрийг дуусах үед “branch_done” event нийтэлж, нэг subscriber терминалд явцыг хэвлэж, нөгөө нь дутуу салбарын тухай анхааруулга бичдэг. Сонирхвол Хичээл 15.1-ийн as_completed урсгалаар collect_all-ийн хувилбар бичиж, салбарын үр дүнг бэлэн болмогц тайланд нэмж болно. Мөн timeout ба retry-ийн тоог тохиргооны файлаас уншиж, Хэмжилтийн хэсгийн бодлогын асуултыг өөр өөр утгаар хэмжих нь энэ төслийг бодит хэрэгсэл болгох эхний алхам. Эдгээр бүгд санал бөгөөд энэ бол сургалтын төгсгөл тул хэн ч таныг тэдгээрийг хийсэн гэж үзэхгүй, харин хийвэл таны өөрийн ажил болно.

Түгээмэл алдаанууд

ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception). “TaskGroup-д барьагдаагүй алдаа” гэсэн утгатай, доор нь жинхэнэ шалтгаан, жишээлбэл KeyError: ‘orders’, гарна. Энэ төсөлд үүнийг санаатайгаар зөвшөөрсөн: кодын алдаа салбарын асуудал мэт нуугдах ёсгүй. Traceback-ийн дотоод хэсгийг уншиж, сервер хүлээгдээгүй хэлбэрийн хариу өгсөн үү, эсвэл таны код буруу уу гэдгийг тодорхойл. Хэрэв серверийн хариуны хэлбэр бодитоор өөрчлөгдсөн бол түүнийг CollectError болгон хөрвүүлэх нь зөв засвар, except Exception биш.

Тест арван секунд гацах. Мессежгүй. make_transport-д slow_branch-ийг хоосон болгохоо мартаж, timeout-ийг багасгаагүй тест Дарханы арван секундийн гацалтад нэг секунд, retry-ийн гурван удаа хүлээнэ. Тест удаан болсон бол хуурамч серверийн тохиргоо ба timeout аргументыг шалга.

error: Missing return statement [return]. “return мөр дутуу” гэсэн утгатай mypy алдаа. retry-ийн wrapper-ийг for attempt in range(1, times + 1): хэлбэрээр бичихэд гарна: mypy давталт return эсвэл raise-гүйгээр дуусаж болно гэж үзнэ. while True хэлбэр энэ асуудлыг арилгадаг.

Нийтлэг rng-ээс болж үр дүн ажиллуулах бүрт өөр байх. Мессежгүй, Алхам 1-д тайлбарласан. Хуурамч сервер бүх хүсэлтэд нэг random.Random хэрэглэвэл зэрэг хүсэлтүүдийн дараалал дүнг өөрчилнө. Тест тогтворгүй болно. Хүсэлт бүрийн өгөгдлийг түүний өөрийн seed-ээс үүсгэ.

Бодит серверт MockTransport-ийн хурдыг хүлээх. Мессежгүй. Хуурамч серверийн 0.2 секундийн тогтмол саатал бодит сүлжээний хэлбэлзлийг загварчилдаггүй: Хичээл 15.2-т Улаанбаатараас нэг хүсэлт 1.3 секунд байсныг санаарай. Бодит серверүүдтэй ажиллахаасаа өмнө тэдгээр дээр timer-ээр дахин хэмжиж, timeout-ийг бодит өгөгдлөөр тохируул.

Товч дүгнэлт

Салбаруудын захиалга цуглуулагч таван түвшний хэрэгслийг нэг програмд нэгтгэсэн. Хуурамч серверүүд MockTransport-ийн async handler-ээр саатал, түр 503 ба хариугүй серверийг тогтвортойгоор загварчилсан. Хүсэлт бүрийн тусдаа seed зэрэг ажиллагаанд ч ижил өгөгдөл баталгаажуулсан. Exception шатлал салбарын асуудлыг, дахин оролдох, бууж өгөх, кодын алдаанаас ялгасан. ParamSpec-тэй async retry decorator 5xx ба timeout-д л дахин оролдсон. asyncio.timeout аль ч транспортод ажилласан. Async generator хуудаслалтыг бүрэн нуусан. Semaphore зэрэг салбарын тоог хязгаарласан, TaskGroup бүх салбарыг хүлээж, кодын алдааг ExceptionGroup-оор гаргасан. Хагас өгөгдөл алдааны шалтгаантай хамт хадгалагдаж, тайлан atomic_write-ээр бичигдсэн. Хэмжилт зэрэгцээ системийн гол хуулийг харуулсан: нийт хугацааг хамгийн удаан ажил тодорхойлдог бөгөөд түүнийг сайжруулах нь timeout ба retry-ийн бодлогын шийдвэр. mypy-ийн strict горим ба долоон тест, үүнд кодын алдаа нуугдахгүй байх ба semaphore-ийн хязгаар, төслийн амлалтуудыг баталгаажуулсан.

Ингээд Python — Гүнзгий түвшин сургалт дууслаа. Та for loop-ийн ард юу байдгаас эхэлж, зэрэг ажилладаг, төрөл нь шалгагдсан, алдаа нь төлөвлөгдсөн, хурд нь хэмжигдсэн програм хүртэл ирлээ. Таван build, тав дахь нь бүгдийг нэгтгэсэн, таны Git түүхэд хадгалагдаж байна: тэдгээр нь таны сурсан зүйлийн нотолгоо, мөн ирээдүйн ажлын загвар.