کلاینت ناهمگام، تلاش دوباره و یکپارچه‌سازی

۱. AsyncClient

httpx.AsyncClient همان رابط Client را دارد، با await. یک کلاینت ناهمگام می‌تواند درون یک نخ درخواست‌های زیادی را هم‌زمان در جریان داشته باشد. درون یک سرور ناهمگام مثل FastAPI، تنها راه درست صدا زدن یک API دیگر هم همین است؛ کلاینت همگام حلقه رویداد را مسدود می‌کند (فصل دوازدهم).

import asyncio

import httpx


async def main() -> None:
    async with httpx.AsyncClient(base_url='http://127.0.0.1:8000', timeout=5) as client:
        page = (await client.get('/courses', params={'sort': 'price'})).json()
        print(f'{page["total"]} courses')
        for course in page['items']:
            print(f'  {course["title"]:<22} {course["price"]:>12,}')

        detail = (await client.get('/courses/3')).json()
        print('course 3 is taught by', detail['instructor']['full_name'])


if __name__ == '__main__':
    try:
        asyncio.run(main())
    except httpx.ConnectError:
        print('Start the session 13 API first.')

AsyncClient در برابر API ما: نمونه 01_async_client/app.py

۲. درخواست‌های هم‌زمان، با سقف

async def sequential(client: httpx.AsyncClient) -> list[str]:
    return [(await client.get(f'/posts/{i}')).json()['title'] for i in IDS]


async def concurrent(client: httpx.AsyncClient) -> list[str]:
    responses = await asyncio.gather(*(client.get(f'/posts/{i}') for i in IDS))
    return [r.json()['title'] for r in responses]


async def concurrent_limited(client: httpx.AsyncClient, limit: int = 5) -> list[str]:
    semaphore = asyncio.Semaphore(limit)

    async def fetch(post_id: int) -> str:
        async with semaphore:  # wait here if `limit` requests are already running
            return (await client.get(f'/posts/{post_id}')).json()['title']

    return list(await asyncio.gather(*(fetch(i) for i in IDS)))

ترتیبی، gather و Semaphore: نمونه 02_concurrent_requests/app.py

sequential           20 posts in 5.90s
concurrent           20 posts in 2.51s
concurrent_limited   20 posts in 0.80s

مؤدب باشید: هرگز ده هزار درخواست را یک‌جا نفرستید. asyncio.Semaphore(5) اجازه می‌دهد حداکثر پنج درخواست هم‌زمان اجرا شوند؛ بسیاری از APIها کلاینت‌هایی را که بیشتر بفرستند مسدود می‌کنند.

۳. تلاش دوباره، با احتیاط

شبکه‌ها و سرورها برای لحظه‌ای شکست می‌خورند: یک مهلت تمام‌شده، یک ۵۰۳ هنگام راه‌اندازی دوباره. تلاش دوباره بعد از مکثی کوتاه اغلب موفق می‌شود. قاعده‌ها:

• فقط درخواست‌های idempotent (GET، PUT، DELETE) دوباره امتحان می‌شوند؛ دو بار انجام دادنشان چیزی را بیشتر تغییر نمی‌دهد. تکرار یک POST می‌تواند دانشجو را دو بار ثبت‌نام کند!
• فقط شکست‌های موقت: خطای انتقال، ۵۰۲، ۵۰۳، ۵۰۴ و ۴۲۹.
• هر بار بیشتر صبر کنید (عقب‌نشینی نمایی): ۰٫۲، ۰٫۴، ۰٫۸ ثانیه.
• بعد از چند تلاش دست بکشید و Retry-After سرور را رعایت کنید.

async def request_with_retry(
    client: httpx.AsyncClient, method: str, url: str, *, attempts: int = 4, backoff: float = 0.2
) -> httpx.Response:
    tries = attempts if method in IDEMPOTENT else 1
    for attempt in range(1, tries + 1):
        try:
            response = await client.request(method, url)
            if response.status_code not in TEMPORARY or attempt == tries:
                return response
            wait = float(response.headers.get('Retry-After', backoff * 2 ** (attempt - 1)))
            print(f'   attempt {attempt}: HTTP {response.status_code}, waiting {wait:.1f}s')
        except httpx.TransportError as error:
            if attempt == tries:
                raise
            wait = backoff * 2 ** (attempt - 1)
            print(f'   attempt {attempt}: {type(error).__name__}, waiting {wait:.1f}s')
        await asyncio.sleep(wait)
    raise AssertionError('unreachable')


def flaky_server() -> httpx.MockTransport:
    calls = {'n': 0}

    def handler(request: httpx.Request) -> httpx.Response:
        calls['n'] += 1
        if calls['n'] == 1:
            raise httpx.ConnectTimeout('slow network', request=request)
        if calls['n'] == 2:
            return httpx.Response(503, headers={'Retry-After': '0.3'})
        return httpx.Response(200, json={'ok': True, 'after_calls': calls['n']})

    return httpx.MockTransport(handler)

تلاش دوباره و یک سرور ناپایدار جعلی: نمونه 03_retry_and_backoff/app.py

GET on a flaky server:
   attempt 1: ConnectTimeout, waiting 0.2s
   attempt 2: HTTP 503, waiting 0.3s
   -> 200 {'ok': True, 'after_calls': 3} in 0.5s
POST on a flaky server (no retry!):
   -> gave up at once: ConnectTimeout

۴. نگاشت خطا و event hook

جدول نگاشت فقط در یک جا زندگی می‌کند و event hookها توابعی‌اند که HTTPX برای هر درخواست و پاسخ صدا می‌زند؛ جای خوبی برای ثبت لاگ، بدون دست زدن به هیچ متدی. نمونه ۰۴ هر دو را با یک سرور جعلی نشان می‌دهد:

  --> GET /admin
  <-- 403 /admin
      PermissionDenied: Admins only
  --> GET /offline
      ApiUnavailable: refused

۵. API شخص ثالث: یکپارچه‌سازی با چیزی که در اختیارتان نیست

class ExternalModel(BaseModel):
    model_config = ConfigDict(extra='ignore')


class User(ExternalModel):
    id: int
    name: str
    email: str


class Todo(ExternalModel):
    userId: int  # noqa: N815  (their field name, not ours)
    completed: bool


class ExternalApiError(Exception):
    pass


async def fetch_list[T: ExternalModel](
    client: httpx.AsyncClient, path: str, model: type[T]
) -> list[T]:
    try:
        response = await client.get(path)
        response.raise_for_status()
        return [model.model_validate(item) for item in response.json()]
    except httpx.HTTPError as error:
        raise ExternalApiError(f'{path}: {error!r}') from error
    except ValidationError as error:
        raise ExternalApiError(
            f'{path}: unexpected data ({error.error_count()} problems)'
        ) from error

مدل‌های حداقلی، مهلت، بررسی و جداسازی: نمونه 05_third_party_api/app.py

• مستندات را بخوانید و فقط فیلدهایی را مدل کنید که لازم دارید (extra='ignore').
• مهلت زمانی همه‌جا؛ از کار افتادن آنها نباید از کار افتادن شما باشد.
• داده آنها را بررسی کنید: فیلد جاافتاده باید یک خطای روشن باشد، نه یک KeyError در عمق کد.
• یکپارچه‌سازی را در یک ماژول و پشت توابع خودتان نگه دارید.

۶. همان API از مرورگر

پوشه 06_browser_fetch یک صفحه HTML است که با جاوااسکریپت و fetch وارد می‌شود، توکن را نگه می‌دارد و دوره‌ها و ثبت‌نام‌های کاربر را می‌خواند. ورود دقیقاً مثل Bruno و HTTPX یک فرم می‌فرستد. سرور بین مرورگر، کلاینت پایتونی و Bruno هیچ فرقی نمی‌گذارد؛ این یعنی یک API خوب طراحی‌شده. برای اینکه CORS اجازه دهد، صفحه را از درگاه ۵۵۰۰ اجرا کنید: python -m http.server 5500.

۷. پروژه: برنامه داشبورد مدرس

07_project/
├── client_app/
│   ├── api_client.py    TrainingCenterApi: توکن، نگاشت خطا، تلاش دوباره
│   ├── models.py        فقط مدل‌هایی که این برنامه لازم دارد
│   └── main.py          داشبورد مدرس با argparse
└── tests/               نه آزمون ناهمگام با MockTransport

    async def _request(self, method: str, path: str, **kwargs: Any) -> httpx.Response:
        headers = kwargs.pop('headers', {})
        if self.token:
            headers['Authorization'] = f'Bearer {self.token}'
        attempts = 1 + (self.retries if method in IDEMPOTENT else 0)
        for attempt in range(1, attempts + 1):
            try:
                response = await self._http.request(method, path, headers=headers, **kwargs)
            except httpx.TransportError as error:
                if attempt == attempts:
                    raise ApiUnavailable(f'{method} {path}: {error!r}') from error
            else:
                if response.status_code not in TEMPORARY or attempt == attempts:
                    if response.is_error:
                        raise _to_error(response)
                    return response
            await asyncio.sleep(self.backoff * 2 ** (attempt - 1))
        raise AssertionError('unreachable')

یک جا برای توکن، تلاش دوباره و خطا: client_app/api_client.py

    async def students_of_many(
        self, course_ids: list[int], limit: int = 5
    ) -> dict[int, list[Student]]:
        """Fetch the students of several courses concurrently (at most `limit` at a time)."""
        semaphore = asyncio.Semaphore(limit)

        async def one(course_id: int) -> tuple[int, list[Student]]:
            async with semaphore:
                return course_id, await self.students(course_id)

        return dict(await asyncio.gather(*(one(cid) for cid in course_ids)))

دانشجویان چند دوره، هم‌زمان و با سقف: client_app/api_client.py

async def dashboard(args: argparse.Namespace) -> None:
    async with TrainingCenterApi(args.base_url) as api:
        await api.login(args.email, args.password)
        me = await api.me()
        print(f'Hello {me.full_name} ({me.role})\n')

        courses = [course async for course in api.all_courses(instructor_id=me.id, sort='title')]
        students = await api.students_of_many([course.id for course in courses])
        print(f'{"course":<24}{"price":>12}{"seats":>9}')
        for course in courses:
            active = sum(s.status == 'active' for s in students[course.id])
            print(f'{course.title:<24}{course.price:>12,}{active:>5}/{course.capacity:<3}')

        if args.new_course:
            try:
                course = await api.create_course(
                    title=args.new_course, price=2_000_000, capacity=12
                )
                course = await api.update_course(course.id, price=2_200_000)
                print(f'\ncreated #{course.id} {course.title!r} for {course.price:,} Toman')
            except Conflict as error:
                print(f'\nnot created: {error.detail}')


def main() -> int:
    parser = argparse.ArgumentParser(description='Teacher dashboard for the Training Center API')
    parser.add_argument('--base-url', default='http://127.0.0.1:8000')
    parser.add_argument('--email', default='teacher@kadoos.ir')
    parser.add_argument('--password', default='Password123')
    parser.add_argument('--new-course', default=None, help='title of a course to create')
    args = parser.parse_args()
    try:
        asyncio.run(dashboard(args))
    except ApiUnavailable:
        print('The API does not answer. Is the server running?', file=sys.stderr)
    except AuthenticationFailed:
        print('Wrong e-mail or password.', file=sys.stderr)
    except PermissionDenied as error:
        print(f'Not allowed: {error.detail}', file=sys.stderr)
    except ApiError as error:
        print(f'The API refused the request: {error}', file=sys.stderr)
    else:
        return 0
    return 1

داشبورد و پیام روشن برای هر شکست: client_app/main.py

Hello Ali Teacher (instructor)

course                         price    seats
Advanced Python            3,500,000    0/15
FastAPI                    4,800,000    0/2
Python Basics              2,500,000    1/20

created #7 'Git Basics' for 2,200,000 Toman

python -m client_app.main --password wrong
Wrong e-mail or password.
python -m client_app.main --base-url http://127.0.0.1:8999
The API does not answer. Is the server running?

۸. خلاصه فصل