User Guide¶
Client¶
from basst import Client
api = Client(
"https://api.example.com",
headers={"X-Token": "secret"},
auth=("user", "pass"),
timeout=5.0,
)
| Parameter | Type | Description |
|---|---|---|
base_url |
str |
Base URL for all requests |
headers |
dict[str, str] \| None |
Default headers |
auth |
tuple[str, str] \| httpx.Auth \| None |
Default auth |
timeout |
float \| None |
Default timeout in seconds |
from_httpx¶
Wrap an existing httpx client. You own the httpx client and must close it yourself.
import httpx
httpx_client = httpx.Client(base_url="https://api.example.com", http2=True)
api = Client.from_httpx(httpx_client)
# ...
httpx_client.close()
HTTP Verbs¶
.get(), .post(), .put(), .patch(), .delete(), .head(), .options() -- each returns a new request builder.
resp = api.get("/users/1").expect()
resp = api.post("/users").json({"name": "Ada"}).expect()
Request Builder¶
All builder methods return self for chaining.
| Method | Description |
|---|---|
.header(name, value) |
Set a single header |
.headers(dict) |
Set multiple headers |
.query(**params) |
Set query parameters |
.cookie(name, value) |
Set a cookie |
.json(data) |
JSON request body |
.data(dict) |
Form-encoded request body |
.auth(auth) |
Per-request auth (None to disable) |
.timeout(seconds) |
Per-request timeout (None to disable) |
.expect() |
Send the request, return Response |
resp = (
api.post("/users")
.header("X-Trace", "abc")
.json({"name": "Ada"})
.timeout(10.0)
.expect()
)
Response Assertions¶
Assertion methods are chainable and raise on failure.
.status(code)¶
Assert the HTTP status code. Raises StatusError.
resp.status(200)
resp.status(201)
.header(name) / .header(name, matcher)¶
Assert a header exists, optionally matching a value. Raises HeaderError.
resp.header("x-request-id") # exists
resp.header("content-type", matchers.contains("json")) # exists + matches
Chaining¶
user = (
resp
.status(200)
.header("content-type", matchers.contains("json"))
.header("x-request-id")
.model(User)
)
Model Binding¶
.model() and .model_list() validate content-type, parse JSON, and return Pydantic model instances.
user = resp.model(User)
users = resp.model_list(User)
Content-Type Validation¶
By default, application/json and any +json suffix are accepted. Controlled via the content_type parameter:
| Value | Behavior |
|---|---|
| (default) | Accept application/json or *+json |
None |
Skip validation |
"problem" |
Prefix match: application/problem+json |
"application/vnd.api+json" |
Exact match |
user = resp.model(User) # default
user = resp.model(User, content_type=None) # skip
problem = resp.model(ProblemDetail, content_type="problem") # prefix
data = resp.model(MyModel, content_type="application/vnd.api+json") # exact
Matchers¶
Built-in matchers for header assertions:
from basst import matchers
matchers.equals("application/json")
matchers.contains("json")
matchers.starts_with("application/")
matchers.ends_with("+json")
matchers.matches(r"^application/(json|.+\+json)$") # regex
A matcher is any Callable[[str], None] that raises AssertionError on failure:
import uuid
def is_uuid(value: str) -> None:
try:
uuid.UUID(value)
except ValueError:
raise AssertionError(f"Expected UUID, got: {value}")
resp.header("x-request-id", is_uuid)
Raw Response Access¶
| Accessor | Type | Description |
|---|---|---|
.status_code |
int |
HTTP status code |
.get_header(name) |
str |
Single header value (raises HeaderError if missing) |
.get_headers() |
httpx.Headers |
All response headers |
.json() |
Any |
Parsed JSON body |
.text |
str |
Raw response body |
.raw |
httpx.Response |
Underlying httpx response (escape hatch) |
Async¶
AsyncClient mirrors Client. The only difference: .expect() is async. Everything after it is synchronous.
from basst import AsyncClient
async with AsyncClient("https://api.example.com") as api:
resp = await api.get("/users/1").expect()
user = resp.status(200).model(User) # no await needed
AsyncClient.from_httpx works the same way:
import httpx
async_httpx_client = httpx.AsyncClient(base_url="https://api.example.com")
aapi = AsyncClient.from_httpx(async_httpx_client)
resp = await aapi.get("/users/1").expect()
await async_httpx_client.aclose()
Context Managers¶
# Sync
with Client("https://api.example.com") as api:
user = api.get("/users/1").expect().status(200).model(User)
# Async
async with AsyncClient("https://api.example.com") as api:
resp = await api.get("/users/1").expect()
user = resp.status(200).model(User)
from_httpx clients are not closed by the context manager -- you own the httpx client.
Errors¶
All errors inherit from BasstError and include HTTP method, URL, and relevant context.
| Error | Raised by | Cause |
|---|---|---|
StatusError |
.status() |
Status code mismatch |
HeaderError |
.header(), .get_header() |
Missing header or matcher failure |
ContentTypeError |
.model(), .model_list() |
Unexpected content-type |
ModelError |
.model(), .model_list() |
Pydantic validation failure |
JsonError |
.json() |
JSON parse failure |
Example error output:
basst.StatusError: Expected status 200, got 404
GET https://api.example.com/users/999
Response body: {"detail": "Not found"}
Out of Scope¶
These will not be supported:
- Non-JSON response binding (XML, HTML, Protobuf, MessagePack)
- JSONPath or string-selector assertions
- Streaming responses (SSE, chunked iteration)
- WebSocket support
- Response body assertions beyond model binding