ساختن یک کلاینت REST پایتونی با HTTPX

۱. از آن سوی میز

کلاینت هر برنامه‌ای است که درخواست می‌فرستد: مرورگر، برنامه موبایل، Bruno یا یک اسکریپت پایتون. HTTPX برای کلاینت همان کاری را می‌کند که FastAPI برای سرور.

۲. پنج روش از سمت کلاینت

import httpx

BASE_URL = 'https://jsonplaceholder.typicode.com'


def main() -> None:
    with httpx.Client(base_url=BASE_URL, timeout=10) as client:
        response = client.get('/posts/1')
        print('GET    ', response.status_code, response.json()['title'][:40])

        response = client.post('/posts', json={'title': 'FastAPI', 'body': 'course', 'userId': 1})
        print('POST   ', response.status_code, response.json())

        response = client.put('/posts/1', json={'id': 1, 'title': 'new', 'body': 'b', 'userId': 1})
        print('PUT    ', response.status_code, response.json()['title'])

        response = client.patch('/posts/1', json={'title': 'only the title'})
        print('PATCH  ', response.status_code, response.json()['title'])

        response = client.delete('/posts/1')
        print('DELETE ', response.status_code)


if __name__ == '__main__':
    try:
        main()
    except httpx.TransportError as error:
        print(f'No internet? {error!r}')

GET، POST، PUT، PATCH، DELETE: نمونه 01_httpx_basics/app.py

httpx.get(url) برای هر درخواست یک اتصال باز و بسته می‌کند. with httpx.Client(base_url=...) یک نشست است: یک مخزن اتصال، با تنظیمات مشترک برای همه درخواست‌ها.

۳. پارامتر، هدر، JSON و فرم

import httpx


def main() -> None:
    with httpx.Client(
        base_url='https://jsonplaceholder.typicode.com',
        headers={'User-Agent': 'kadoos-course-client/1.0'},  # sent with EVERY request
        timeout=10,
    ) as client:
        response = client.get('/comments', params={'postId': 1, '_limit': 3})
        print(response.request.url)
        print(response.status_code, response.headers['content-type'])
        print(f'took {response.elapsed.total_seconds() * 1000:.0f} ms')
        for comment in response.json():
            print(' -', comment['email'])

        response = client.post('/posts', json={'title': 'Hello', 'body': 'from httpx', 'userId': 7})
        print(response.request.headers['content-type'], '->', response.json())

        form = client.post('/posts', data={'title': 'as a form'})
        print(form.request.headers['content-type'])


if __name__ == '__main__':
    try:
        main()
    except httpx.TransportError as error:
        print(f'No internet? {error!r}')

ساختن درخواست و خواندن پاسخ: نمونه 02_params_headers_json/app.py

۴. مهلت زمانی و خطاها

یک کلاینت باید انتظار شکست را داشته باشد. دو نوع شکست کاملاً متفاوت داریم:

1. هیچ پاسخی نیامد: HTTPX استثنا می‌دهد. httpx.ConnectError یعنی سرور خاموش است یا نشانی اشتباه است و httpx.TimeoutException یعنی سرور بیش از حد کند است. هر دو زیرکلاس httpx.TransportError هستند.
2. پاسخی با کد خطا آمد: به‌طور پیش‌فرض استثنا نیست. response.is_success را بررسی کنید یا raise_for_status() بزنید.

import httpx


def fake_server(request: httpx.Request) -> httpx.Response:
    if request.url.path == '/slow':
        raise httpx.ReadTimeout('the server did not answer in time', request=request)
    if request.url.path == '/down':
        raise httpx.ConnectError('connection refused', request=request)
    if request.url.path == '/courses/1':
        return httpx.Response(200, json={'id': 1, 'title': 'FastAPI'})
    if request.url.path == '/broken':
        return httpx.Response(500, text='Internal Server Error')
    return httpx.Response(404, json={'detail': 'Not found'})


def fetch(client: httpx.Client, path: str) -> None:
    try:
        response = client.get(path)
        response.raise_for_status()
    except httpx.TimeoutException:
        print(f'{path:<12} timeout        -> try again later')
    except httpx.ConnectError:
        print(f'{path:<12} no connection  -> is the server running?')
    except httpx.HTTPStatusError as error:
        status = error.response.status_code
        kind = 'our mistake (4xx)' if status < 500 else 'server problem (5xx)'
        print(f'{path:<12} HTTP {status}       -> {kind}')
    else:
        print(f'{path:<12} OK             -> {response.json()}')


def main() -> None:
    timeout = httpx.Timeout(5.0, connect=2.0)  # 2 s to connect, 5 s for the rest
    with httpx.Client(
        base_url='http://api.test', transport=httpx.MockTransport(fake_server), timeout=timeout
    ) as client:
        for path in ['/courses/1', '/courses/99', '/broken', '/slow', '/down']:
            fetch(client, path)


if __name__ == '__main__':
    main()

همه نوع شکست، شبیه‌سازی‌شده با MockTransport: نمونه 03_timeouts_and_errors/app.py

/courses/1   OK             -> {'id': 1, 'title': 'FastAPI'}
/courses/99  HTTP 404       -> our mistake (4xx)
/broken      HTTP 500       -> server problem (5xx)
/slow        timeout        -> try again later
/down        no connection  -> is the server running?

httpx.MockTransport پاسخ‌های جعلی برمی‌گرداند، پس این مثال به سرور و اینترنت نیاز ندارد. آزمون‌های کلاینت همین ترفند را استفاده می‌کنند.

۵. ورود و توکن Bearer از سمت کلاینت

class BearerAuth(httpx.Auth):
    """Adds the Authorization header to every request of a client."""

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

    def auth_flow(self, request: httpx.Request):  # type: ignore[no-untyped-def]
        request.headers['Authorization'] = f'Bearer {self.token}'
        yield request


def main() -> None:
    with httpx.Client(base_url=API, timeout=5) as client:
        print('without token:', client.get('/users/me').status_code)

        login = client.post(
            '/auth/token', data={'username': 'sara@example.com', 'password': 'Password123'}
        )
        login.raise_for_status()
        token = login.json()['access_token']
        print('token starts with', token[:20], '...')

        me = client.get('/users/me', headers={'Authorization': f'Bearer {token}'})
        print('by hand      :', me.json()['full_name'])

    with httpx.Client(base_url=API, timeout=5, auth=BearerAuth(token)) as client:
        print('with Auth    :', client.get('/users/me').json()['email'])
        print(
            'my courses   :',
            [e['course']['title'] for e in client.get('/users/me/enrollments').json()],
        )

فرم ورود، ذخیره توکن و httpx.Auth: نمونه 04_bearer_auth/app.py

یک کلاس کوچک httpx.Auth هدر مجوز را یک بار برای کل کلاینت اضافه می‌کند. پروژه از همین شیوه استفاده می‌کند.

۶. از درخواست‌های پراکنده تا یک کلاس

اسکریپت‌هایی پر از httpx.get(f'{API}/courses/{id}') نشانی پایه، هدرها و بررسی خطا را همه‌جا تکرار می‌کنند. یک کلاس کلاینت همه این‌ها را در یک جا نگه می‌دارد و متدهایی با نام‌های کسب‌وکار ارائه می‌دهد که شیء تایپ‌دار برمی‌گردانند، نه دیکشنری خام:

@dataclass(frozen=True)
class Course:
    id: int
    title: str
    price: int


class CoursesClient:
    def __init__(self, base_url: str, timeout: float = 5.0) -> None:
        self._http = httpx.Client(base_url=base_url, timeout=timeout)

    # context manager: `with CoursesClient(...) as api:` closes the connections
    def __enter__(self) -> Self:
        return self

    def __exit__(self, *exc: type[BaseException] | BaseException | TracebackType | None) -> None:
        self._http.close()

    def list_courses(self, q: str | None = None) -> list[Course]:
        params = {'q': q} if q else {}
        response = self._http.get('/courses', params=params)
        response.raise_for_status()
        return [Course(c['id'], c['title'], c['price']) for c in response.json()['items']]

    def get_course(self, course_id: int) -> Course | None:
        response = self._http.get(f'/courses/{course_id}')
        if response.status_code == 404:
            return None
        response.raise_for_status()
        data = response.json()
        return Course(data['id'], data['title'], data['price'])

نسخه اول کلاس: نمونه 05_client_class/app.py

۷. پروژه: کلاینت آموزشگاه

06_project/
├── clients/
│   ├── errors.py                    استثناهای کلاینت و نگاشت کد وضعیت
│   ├── models.py                    مدل‌های تایپ‌دار JSON، مال خود کلاینت
│   └── training_center_client.py    کلاس TrainingCenterClient
├── demo.py                          استفاده واقعی در برابر سرور فصل ۱۳
└── tests/test_client.py             چهارده آزمون با MockTransport

    def _request(self, method: str, path: str, **kwargs: Any) -> httpx.Response:
        try:
            response = self._http.request(method, path, **kwargs)
        except httpx.TimeoutException as error:
            raise ApiConnectionError(f'{method} {path} timed out') from error
        except httpx.TransportError as error:
            raise ApiConnectionError(f'{method} {path} failed: {error}') from error
        if response.is_error:
            raise error_from_response(response)
        return response

    # --- authentication ------------------------------------------------
    def register(self, email: str, full_name: str, password: str) -> User:
        body = {'email': email, 'full_name': full_name, 'password': password}
        return User.model_validate(self._request('POST', '/auth/register', json=body).json())

    def login(self, email: str, password: str) -> Token:
        """OAuth2 password flow: a FORM, not JSON. The token is kept for later calls."""
        response = self._request(
            'POST', '/auth/token', data={'username': email, 'password': password}
        )
        token = Token.model_validate(response.json())
        self.token = token.access_token
        return token

    def logout(self) -> None:
        """JWTs live on the client: logging out = forgetting the token."""
        self.token = None

    def me(self) -> User:
        return User.model_validate(self._request('GET', '/users/me').json())

یک جا برای فرستادن و خطا، و ورود: clients/training_center_client.py

    def iter_courses(self, *, size: int = 50, **filters: Any) -> Iterator[Course]:
        """A generator over EVERY page: callers never think about pagination."""
        page = 1
        while True:
            result = self.list_courses(page=page, size=size, **filters)
            yield from result.items
            if page >= result.pages:
                return
            page += 1

مولدی که صفحه‌بندی را پنهان می‌کند: clients/training_center_client.py

_STATUS_TO_ERROR: dict[int, type[ApiError]] = {
    401: AuthenticationFailed,
    403: PermissionDenied,
    404: NotFound,
    409: ConflictError,
    422: ValidationFailed,
}


def error_from_response(response: httpx.Response) -> ApiError:
    """Build the most specific `ApiError` subclass for an error response."""
    try:
        body = response.json()
    except ValueError:
        body = {'detail': response.text or response.reason_phrase}
    detail = body.get('detail', body) if isinstance(body, dict) else body
    code = body.get('code') if isinstance(body, dict) else None
    status = response.status_code
    error_class = _STATUS_TO_ERROR.get(status, ServerError if status >= 500 else ApiError)
    return error_class(
        status, detail, code, method=response.request.method, url=str(response.request.url)
    )

نگاشت کد وضعیت به استثنا: clients/errors.py

• یک httpx.Client با نشانی پایه، مهلت زمانی و User-Agent.
• توکن بعد از login() درون کلاینت می‌ماند و با هر درخواست فرستاده می‌شود؛ logout() یعنی فراموش کردن توکن.
• هر متد یک مدل تایپ‌دار برمی‌گرداند و هر شکست به یکی از استثناهای خود کلاینت تبدیل می‌شود.
• مدل‌های کلاینت عمداً از app.schemas سرور وارد نمی‌شوند؛ یک کلاینت واقعی در مخزن دیگری زندگی می‌کند و فقط قرارداد JSON را می‌شناسد. فیلدهای ناشناخته نادیده گرفته می‌شوند تا سرور تازه‌تر کلاینت قدیمی را خراب نکند.

Registered and logged in as Demo Student (student)
Enrolled in course 1: active
Second time -> 409: Already enrolled
Student creates a course -> 403: This action requires the role: admin or instructor
Teacher created #6 and changed its price to 1,200,000

۸. خلاصه فصل