RESTful Mixin¶

def auth_flow(self, request: httpx.Request) -> t.Generator[httpx.Request, httpx.Response, None]:
    """Add Bearer token to request Authorization header.

    Args:
        request: The HTTP request to authenticate.

    Yields:
        The authenticated request with Authorization header.
    """
    request.headers["Authorization"] = f"Bearer {self.token}"
    yield request

async def request(
    self,
    endpoint: str,
    *,
    method: t.Literal["GET", "POST", "PUT", "DELETE"] = "GET",
    headers: t.Mapping[str, str] | None = None,
    params: t.Mapping[str, t.Any] | None = None,
    json: t.Mapping[str, t.Any] | None = None,
    cast_to: type[RespT],
    validate: bool = True,
    strict: bool = True,
    raise_for_status: bool = True,
    max_retries: int = 3,
    retry_wait: float = 2.0,
) -> RespT:
    """Make an HTTP request with automatic retry and response
    validation.

    Sends an HTTP request to the specified endpoint and parses the response
    into the specified Pydantic model. Automatically retries on failure and
    validates the response data.

    Args:
        endpoint: API endpoint path (relative to base_url).
        method: HTTP method to use. Defaults to "GET".
        headers: Optional request headers to merge with default headers.
        params: Optional query parameters to merge with default params.
        json: Optional JSON body for POST/PUT requests.
        cast_to: Pydantic model class to parse the response into.
        validate: Whether to validate response data. Defaults to True.
        strict: Whether to use strict validation (reject extra fields).
            Defaults to True.
        raise_for_status: Whether to raise an exception for HTTP error
            status codes. Defaults to True.
        max_retries: Maximum number of retry attempts. Defaults to 3.
        retry_wait: Wait time in seconds between retries. Defaults to 2.0.

    Returns:
        Parsed response as an instance of cast_to model.

    Raises:
        httpx.HTTPStatusError: If the response status indicates an error
            after all retries.
        pyd.ValidationError: If response validation fails.
        tenacity.RetryError: If all retry attempts are exhausted.

    Example:
        ```python
        class UserResponse(pyd.BaseModel):
            user_id: str
            username: str
            email: str


        # GET request
        user = await api.request(
            "/users/123", method="GET", cast_to=UserResponse
        )

        # POST request with retry
        new_user = await api.request(
            "/users",
            method="POST",
            json={"username": "john", "email": "john@example.com"},
            cast_to=UserResponse,
            max_retries=5,
            retry_wait=3.0,
        )
        ```
    """

    @tenacity.retry(
        stop=tenacity.stop_after_attempt(max_retries),
        wait=tenacity.wait_fixed(retry_wait),
        reraise=True,
    )
    async def _do_request(
        method: str,
        endpoint: str,
        headers: t.Mapping[str, str] | None,
        params: t.Mapping[str, t.Any] | None,
        json: t.Mapping[str, t.Any] | None,
    ) -> httpx.Response:
        response = await self.http_client.request(
            method,
            endpoint,
            headers=headers,
            params=params,
            json=json,
        )
        if raise_for_status:
            response.raise_for_status()
        return response

    response = await _do_request(method, endpoint, headers, params, json)
    if validate:
        return cast_to.model_validate(response.json(), strict=strict)
    return cast_to.model_construct(**response.json())

async def close(self) -> None:
    """Close the HTTP client and cleanup resources.

    Should be called when the API client is no longer needed to properly
    close all connections. Automatically called when using as a context manager.

    Example:
        ```python
        api = MyAPI("https://api.example.com")
        try:
            await api.get_data()
        finally:
            await api.close()

        # Or use as context manager
        async with MyAPI("https://api.example.com") as api:
            await api.get_data()
        ```
    """
    await self.http_client.aclose()