soft-http 0.1.0

Async Python HTTP client

soft-http

Async Python HTTP client

Highlights

• Typed generic responses — client.get(url, response_type=User) deserializes JSON directly into your dataclass or Pydantic model; no Any in the public API
• First-class hooks — before_request, after_response, before_retry, before_error; just append an async callable
• Built-in retry with exponential backoff — one line: retry=RetryConfig(limit=3)
• Instance config — set prefix_url, headers, timeout, and auth once on the client, override per-request
• Zero-config deserialization — auto-detects Pydantic models and dataclasses, no serializer config needed
• Fully typed, ships py.typed, works great with mypy and pyright
• Async-first — built on aiohttp

Install

pip install soft-http

uv add soft-http

Usage

import asyncio
from dataclasses import dataclass
from soft_http import SoftClient, ClientConfig


@dataclass
class User:
    id: int
    name: str


async def main() -> None:
    config = ClientConfig(prefix_url="https://jsonplaceholder.typicode.com")

    async with SoftClient(config=config) as client:
        response = await client.get("users/1", response_type=User)
        print(response.data)  # User(id=1, name='Leanne Graham')


asyncio.run(main())

API

SoftClient(config=None, *, session=None)

The main entry point. Every instance holds its own hook registry and an optional base config.

from soft_http import SoftClient, ClientConfig

client = SoftClient(config=ClientConfig(prefix_url="https://api.example.com"))

# Use as an async context manager (recommended):
async with SoftClient(config=config) as client:
    ...

# Or manage the lifecycle manually:
client = SoftClient(config=config)
try:
    ...
finally:
    await client.close()

Injecting an existing session — useful in frameworks like FastAPI where the session has its own lifespan:

import aiohttp

async with aiohttp.ClientSession() as session:
    client = SoftClient(config=config, session=session)
    # SoftClient won't close the injected session

HTTP methods

All methods are async and return ClientResponse[T] (or ClientResponse[bytes] when response_type is omitted).

await client.get(url, *, response_type=None, config=None)
await client.post(url, *, response_type=None, data=None, config=None)
await client.put(url, *, response_type=None, data=None, config=None)
await client.patch(url, *, response_type=None, data=None, config=None)
await client.delete(url, *, response_type=None, config=None)

Parameter	Type	Description
url	str	Path appended to prefix_url, or a full URL. Must not start with / when prefix_url is set
response_type	type[T] | None	Target class for JSON deserialization; omit to get raw bytes
data	Any	Request body passed to aiohttp. For a JSON body, pass a JSON string and set the Content-Type header
config	ClientConfig | None	Per-request overrides, deep-merged with instance config

client.hooks

A Hooks instance. Append async callables to any of its four lists to intercept the request lifecycle.

---

ClientConfig

A dataclass for configuring client instances and individual requests.

from soft_http import ClientConfig, RetryConfig

config = ClientConfig(
    prefix_url="https://api.example.com",
    headers={"Authorization": "Bearer token"},
    params={"version": "2"},
    timeout=10.0,
    auth=("user", "pass"),
    retry=RetryConfig(limit=3),
    throw_http_errors=True,
)

Field	Type	Default	Description
prefix_url	str	""	Base URL prepended to every request path
headers	dict[str, str]	{}	Default headers; deep-merged across instance and per-request configs
params	SearchParams	None	Query parameters — see SearchParams
timeout	float | None	None	Request timeout in seconds
auth	tuple[str, str] | None	None	HTTP Basic Auth as (username, password)
retry	RetryConfig | int | None	None	Retry policy; pass int as shorthand for RetryConfig(limit=n)
throw_http_errors	bool	True	True raises ClientResponseException on 4xx/5xx; False returns the response

SearchParams

type SearchParams = str | Mapping[str, str] | Sequence[tuple[str, str]] | None

All three formats are equivalent:

"page=1&limit=10"
{"page": "1", "limit": "10"}
[("page", "1"), ("limit", "10")]

Use a list of tuples for duplicate keys:

params=[("tag", "python"), ("tag", "async")]
# → ?tag=python&tag=async

---

RetryConfig

Controls automatic retries with exponential backoff.

from soft_http import RetryConfig

RetryConfig(
    limit=2,
    methods=frozenset({"GET", "POST"}),
    status_codes=frozenset({429, 503}),
    backoff_limit=30.0,
)

Field	Type	Default	Description
limit	int	2	Maximum number of retry attempts
methods	frozenset[str]	{"GET", "PUT", "HEAD", "DELETE", "OPTIONS", "TRACE"}	HTTP methods eligible for retry
status_codes	frozenset[int]	{408, 413, 429, 500, 502, 503, 504}	Response status codes that trigger a retry
backoff_limit	float	30.0	Maximum delay in seconds (delay grows exponentially but is capped here)

Shorthand — pass a plain int to use all defaults with a custom limit:

config = ClientConfig(retry=3)  # RetryConfig(limit=3, …defaults)

---

ClientResponse[T]

The return value of every HTTP method call.

response = await client.get("users/1", response_type=User)

response.status_code   # int, e.g. 200
response.headers       # dict[str, str] — keys are lower-cased
response.data          # T — deserialized User instance
response.raw           # bytes — original response body
response.url           # str — final response URL (after redirects)
response.content_type  # str — e.g. "application/json", without charset

Field / property	Type	Description
status_code	int	HTTP status code
headers	dict[str, str]	Response headers (keys lower-cased)
data	T	Deserialized body when response_type is set, otherwise bytes
raw	bytes	Raw response body, always present
url	str	Final response URL after redirects
content_type	str	content-type header without the charset suffix (property)

Deserialization is automatic:

• Pydantic models — uses Model.model_validate(json_data)
• Dataclasses and plain classes — uses Class(**json_data)

---

Hooks

Each SoftClient exposes a hooks attribute — a simple dataclass with four lists of async callables.

client.hooks.before_request   # list[BeforeRequestHook]
client.hooks.after_response   # list[AfterResponseHook]
client.hooks.before_retry     # list[BeforeRetryHook]
client.hooks.before_error     # list[BeforeErrorHook]

Append to add, remove by reference to deregister:

client.hooks.before_request.append(my_hook)
client.hooks.before_request.remove(my_hook)

Hook signatures

# Called before every request; mutate and return the config
async def before_request_hook(config: ClientConfig) -> ClientConfig: ...

# Called after every successful response; mutate and return the response
async def after_response_hook(response: ClientResponse[Any]) -> ClientResponse[Any]: ...

# Called before each retry attempt
async def before_retry_hook(ctx: RetryContext) -> None: ...

# Called before raising ClientResponseException; mutate and return the error
async def before_error_hook(error: ClientResponseException) -> ClientResponseException: ...

RetryContext

@dataclass
class RetryContext:
    request: ClientConfig
    retry_count: int
    error: ClientResponseException

---

Exceptions

ClientException
├── ClientRequestException   # Network-level failure (connection refused, timeout, …)
└── ClientResponseException  # 4xx or 5xx response
      .status_code: int
      .response: ClientResponse[bytes]
      .request_url: str
      .request_method: str
      .retry_count: int

from soft_http import ClientException, ClientRequestException, ClientResponseException

try:
    response = await client.get("users/999")
except ClientResponseException as e:
    print(e.status_code)       # 404
    print(e.response.raw)      # b'{"error": "not found"}'
except ClientRequestException as e:
    print(f"Network error: {e}")

---

Tips

Add auth with a before_request hook

async def attach_token(config: ClientConfig) -> ClientConfig:
    config.headers["Authorization"] = f"Bearer {get_token()}"
    return config

client.hooks.before_request.append(attach_token)

Log every response

from soft_http import ClientResponse
from typing import Any

async def log_response(response: ClientResponse[Any]) -> ClientResponse[Any]:
    print(f"{response.status_code} — {len(response.raw)} bytes")
    return response

client.hooks.after_response.append(log_response)

Handle errors without exceptions

config = ClientConfig(prefix_url="https://api.example.com", throw_http_errors=False)

async with SoftClient(config=config) as client:
    response = await client.get("might-not-exist")
    if response.status_code == 404:
        print("not found")

Shared session in FastAPI

from contextlib import asynccontextmanager
from fastapi import FastAPI
import aiohttp
from soft_http import SoftClient, ClientConfig

http: SoftClient


@asynccontextmanager
async def lifespan(app: FastAPI):
    global http
    session = aiohttp.ClientSession()
    http = SoftClient(config=ClientConfig(prefix_url="https://api.example.com"), session=session)
    yield
    await session.close()


app = FastAPI(lifespan=lifespan)

Monitor retries

from soft_http import RetryContext

async def on_retry(ctx: RetryContext) -> None:
    print(f"Retry #{ctx.retry_count} after HTTP {ctx.error.status_code}")

client.hooks.before_retry.append(on_retry)

---

FAQ

Why not requests or httpx?

soft-http is async-first and built on top of aiohttp. If you need synchronous HTTP, requests or httpx are better choices. If you need async HTTP with a clean, high-level API, soft-http is for you.

Why not bare aiohttp?

aiohttp is a fantastic low-level library but requires boilerplate for things like retry logic, response deserialization, typed responses, and request/response interceptors. soft-http provides all of that out of the box with zero Any in the public API.

Does it support streaming responses?

Not yet. The current API buffers the full response body. Streaming support is planned.

---

License

MIT © Zero-i00