Skip to content

engine.http — Base async HTTP client

⬅️ Back | 🏠 Docs Root

BaseApiClient

BaseApiClient

Base async HTTP client with a long-lived connection pool.

Created once in the DI container, not per-request. Reuses TCP connections via httpx connection pooling. Call close() when stopping the bot for proper termination.

Parameters:

Name Type Description Default
base_url str

Base API URL (e.g., "https://api.example.com").

 required
api_key str | None

API key for the X-API-Key header. None — no authentication.

 None
timeout float

Total request timeout in seconds (connect is always 5 sec).

 10.0
Example 
class BookingApiClient(BaseApiClient):
    async def get_slots(self, date: str) -> list[dict]:
        return await self._request("GET", "/slots", params={"date": date})

# In the DI container — once:
client = BookingApiClient(base_url="https://api.example.com", api_key="secret")

# When stopping the bot:
await client.close()
Source code in src/codex_bot/engine/http/api_client.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
class BaseApiClient:
    """
    Base async HTTP client with a long-lived connection pool.

    Created once in the DI container, not per-request.
    Reuses TCP connections via httpx connection pooling.
    Call ``close()`` when stopping the bot for proper termination.

    Args:
        base_url: Base API URL (e.g., ``"https://api.example.com"``).
        api_key: API key for the ``X-API-Key`` header. ``None`` — no authentication.
        timeout: Total request timeout in seconds (``connect`` is always 5 sec).

    Example:
        ```python
        class BookingApiClient(BaseApiClient):
            async def get_slots(self, date: str) -> list[dict]:
                return await self._request("GET", "/slots", params={"date": date})

        # In the DI container — once:
        client = BookingApiClient(base_url="https://api.example.com", api_key="secret")

        # When stopping the bot:
        await client.close()
        ```
    """

    def __init__(
        self,
        base_url: str,
        api_key: str | None = None,
        timeout: float = 10.0,
    ) -> None:
        headers: dict[str, str] = {
            "Content-Type": "application/json",
            "Accept": "application/json",
        }
        if api_key:
            headers["X-API-Key"] = api_key

        # Long-lived client — created once, lives for the entire bot duration
        self.client = httpx.AsyncClient(
            base_url=base_url.rstrip("/"),
            headers=headers,
            timeout=httpx.Timeout(timeout, connect=5.0),
        )

    async def close(self) -> None:
        """Closes the connection pool. Call when stopping the bot.

        Example:
            ```python
            # In on_shutdown hook:
            await api_client.close()
            ```
        """
        await self.client.aclose()

    async def _request(
        self,
        method: str,
        endpoint: str,
        json: dict[str, Any] | None = None,
        params: dict[str, Any] | None = None,
    ) -> Any:
        """Performs an HTTP request via the long-lived client.

        Args:
            method: HTTP method (``"GET"``, ``"POST"``, ``"PUT"``, ``"DELETE"``).
            endpoint: Endpoint path (e.g., ``"/api/v1/slots"``).
            json: Request body in JSON format.
            params: Query parameters.

        Returns:
            Parsed JSON response or ``None`` for 204 No Content.

        Raises:
            ApiClientError: For HTTP errors or connection issues.
        """
        url = endpoint.lstrip("/")

        try:
            log.debug(f"API {method} /{url} | params={params}")
            response = await self.client.request(method=method, url=url, json=json, params=params)
            response.raise_for_status()

            # 204 No Content and empty responses (DELETE, some POST)
            if response.status_code == 204 or not response.content:
                return None

            return response.json()

        except httpx.HTTPStatusError as e:
            log.error(f"API HTTP error {e.response.status_code}: {e.response.text}")
            raise ApiClientError(f"HTTP {e.response.status_code}") from e
        except httpx.RequestError as e:
            log.error(f"API connection error: {e}")
            raise ApiClientError(f"Connection error: {e}") from e

Functions

close() async

Closes the connection pool. Call when stopping the bot.

Example 
# In on_shutdown hook:
await api_client.close()
Source code in src/codex_bot/engine/http/api_client.py 
73
74
75
76
77
78
79
80
81
82
async def close(self) -> None:
    """Closes the connection pool. Call when stopping the bot.

    Example:
        ```python
        # In on_shutdown hook:
        await api_client.close()
        ```
    """
    await self.client.aclose()

ApiClientError

ApiClientError

Bases: Exception

Base HTTP client error.

Source code in src/codex_bot/engine/http/api_client.py 
22
23
class ApiClientError(Exception):
    """Base HTTP client error."""