ahttp-client

ahttp-client is a decorator-based HTTP client framework that maps typed function parameters to HTTP requests.

Key Features

• Declare HTTP endpoints using @request or the @get, @post, @put, @patch, @delete, and @options decoration methods.
• Use typing.Annotated to set HTTP parameters such as the path, query, header, or body values.
• Serialize typed request models and deserialize responses using registered codecs.
• Customize the request lifecycle using before_hook and after_hook decorators.
• Reduce boilerplate code when using HTTP client packages such as aiohttp, httpx, and requests.

Installation

Install the extra for the HTTP client library you want to use. Python 3.11 or later is required.

pip install "ahttp-client[aiohttp]"
pip install "ahttp-client[httpx]"
pip install "ahttp-client[requests]"

Include the pydantic extra to serialize and deserialize Pydantic models.

pip install "ahttp-client[aiohttp,pydantic]"

Quick start

Asynchronous Client

Declare a service by extending AsyncSession, then decorate coroutine methods with an HTTP method and path. Annotated parameters determine where values are placed in the request.

import asyncio
from typing import Annotated, Any

import aiohttp

from ahttp_client import AsyncSession, Path, Response, get


class GitHubService(AsyncSession):
    def __init__(self):
        super().__init__("https://api.github.com", aiohttp.ClientSession)

    @get("/users/{user}/repos")
    async def list_repositories(
        self, response: Response, user: Annotated[str, Path]
    ) -> list[dict[str, Any]]:
        return response.json()


async def main():
    async with GitHubService() as service:
        repositories = await service.list_repositories(user="gunyu1019")
        print(repositories)


asyncio.run(main())

AsyncSession closes its underlying HTTP client when the async with block ends. Decorated responses are also closed automatically after the handler returns.

Synchronous Client

Use Session and a regular function with requests.Session or httpx.Client.

from typing import Annotated, Any

import requests

from ahttp_client import Path, Response, Session, get


class GitHubService(Session):
    def __init__(self):
        super().__init__("https://api.github.com", requests.Session)

    @get("/users/{user}/repos")
    def list_repositories(
        self, response: Response, user: Annotated[str, Path]
    ) -> list[dict[str, Any]]:
        return response.json()


with GitHubService() as service:
    repositories = service.list_repositories(user="gunyu1019")
    print(repositories)

Request components

Use Annotated to describe dynamic request values.

from typing import Annotated

from ahttp_client import BodyJson, Header, Path, Query, Response, post


class UserService(AsyncSession):
    @post("/users/{user_id}")
    async def update_user(
        self,
        response: Response,
        user_id: Annotated[int, Path],
        verbose: Annotated[bool, Query],
        authorization: Annotated[str, Header.custom_name("Authorization")],
        display_name: Annotated[str, BodyJson.custom_key("profile.displayName")],
    ) -> dict:
        return response.json()

Set directly_response=True on a request (or a session) when you need the Response object itself instead of running the decorated handler. In that case, close it yourself with await response.async_close() for async clients or response.close() for sync clients.

Model serialization

Registered codecs can convert a complete Body parameter before transport and validate a direct response from its return annotation. When Pydantic is installed, BaseModel types and nested model containers are supported automatically.

Use @serialize and @deserialize to pass codec options. If the model argument is omitted, the request body and return annotations select the codec after the request decorator is applied.

from typing import Annotated

from pydantic import BaseModel

from ahttp_client import AsyncSession, Body, post
from ahttp_client.serializer import deserialize, serialize


class CreateUser(BaseModel):
    name: str
    nickname: str | None = None


class User(BaseModel):
    id: int
    name: str


class UserService(AsyncSession):
    @post("/users", directly_response=True)
    @serialize(exclude_none=True)
    @deserialize(strict=True)
    async def create_user(
        self,
        user: Annotated[CreateUser, Body],
    ) -> User:
        ...

In this example, the request body is produced with BaseModel.model_dump(mode="json", exclude_none=True), and the JSON response is validated as User. Because directly_response=True selects deserialized mode from the registered return type, the decorated method body is not executed. Pass a model explicitly, such as @serialize(CreateUser), when it cannot be inferred from an annotation.

Hooks

Attach a hook to a decorated request to modify it before dispatch or transform its result afterward. Async requests require async hooks; sync requests require regular functions.

class GitHubService(AsyncSession):
    @get("/user")
    async def current_user(self, response: Response) -> dict:
        return response.json()

    @current_user.before_hook
    async def add_authorization(self, request, path):
        request.headers["Authorization"] = "Bearer <token>"
        return request, path

Override before_request() or after_request() on AsyncSession or Session to apply the same behavior to every request in a service.

Documentation

• English documentation
• 한국어 문서