Status & timing

A11's error and time types. Native failures cross the boundary as StatusException with structured details preserved.

Error boundaries

An A11 status contains a canonical code, message, and structured details. Actions, nodes, sessions, stores, and transports retain that status through asynchronous cleanup and remote propagation. Python wait methods and consume() raise StatusException for a non-OK result, preserving the original status for code-based handling.

A unary read reaches success through the node's final marker and OK closure. Streaming iteration can end at a clean close after yielding independent values. An abort exposes the non-OK status in either case. An action's completion status separately reports the operation and its output cleanup.

A status can also travel as application data in a status chunk with media type application/x-a11-status. This representation is used for dispatch, completion, node abort, and closure records. It remains separate from ordinary serialization because a non-OK StatusOr represents an error rather than a value.

Status

a11.status.Status

Status(code: SupportsInt = 0, message: str = 'OK', details: Any = [])

Creates a status from a canonical code, message, and details list.

code property writable

code: StatusCode

The canonical status code.

details property writable

details: list[Any]

The structured status details, as a list.

message property writable

message: str

The human-readable status message.

from_exception staticmethod

from_exception(exc: BaseException, casters: StatusExceptionCasters | None = None) -> Status

Convert an application exception to a transportable status.

Existing StatusException values retain their structured status; registered casters handle framework-specific types, and unknown exceptions become UNKNOWN.

from_http_exception staticmethod

from_http_exception(http_exception: HTTPException | HTTPStatusError) -> Status

Convert a FastAPI/httpx HTTP exception to an A11 status.

get_fastapi_response_dict_for_codes staticmethod

get_fastapi_response_dict_for_codes(*codes: StatusCode) -> dict[int, dict]

Build FastAPI response documentation for portable status codes.

The examples are plain JSON documents rather than Status values. FastAPI serialises the whole OpenAPI document with pydantic, and a native Status inside it is not something pydantic can serialise -- which made app.openapi() fail for any route that documented one.

get_fastapi_response_dict_for_http_codes staticmethod

get_fastapi_response_dict_for_http_codes(*codes: int) -> dict[int, dict]

Build FastAPI response documentation for explicit HTTP codes.

ok staticmethod

ok(message: str | None = None) -> Status

Create a successful status with an optional descriptive message.

parse_from_json staticmethod

parse_from_json(data: str | bytes) -> StatusParseResult

Parse status JSON and return validation state without throwing.

is_ok

is_ok() -> bool

Returns whether the status is OK (no error).

raise_if_not_ok

raise_if_not_ok() -> None

Raise StatusException when this status is non-OK.

to_exception

to_exception() -> StatusException

Convert a non-OK status to its Python boundary exception.

to_msgpack

to_msgpack(packer: Packer) -> None

Append this status to an A11 MessagePack encoder.

a11.status.StatusCode

Bases: IntEnum

Portable gRPC/Abseil outcome codes used throughout A11.

Pick the most specific code a caller can act on: for example, INVALID_ARGUMENT for input that can never work, FAILED_PRECONDITION for a lifecycle state that could change, and UNAVAILABLE for a dependency whose operation may succeed on retry. The conversion methods preserve that intent at HTTP and WebSocket boundaries.

from_http_code staticmethod

from_http_code(http_code: int) -> StatusCode

Map an HTTP response code to the nearest portable status code.

from_ws_code staticmethod

from_ws_code(code: int) -> StatusCode

Decode standard and A11-private WebSocket close codes.

to_ws_code

to_ws_code() -> int

Encode this outcome as a WebSocket close code.

to_http_code

to_http_code() -> int

Map this outcome to the corresponding HTTP response code.

a11.status.StatusException

StatusException(status: Status)

Bases: Exception

Python exception carrying a structured non-OK A11 status.

Catch this at an action or service boundary when you want to inspect or forward status.code, status.message, and status.details. A11's native Python bindings raise this type consistently rather than exposing an Abseil or pybind11-specific exception.

Timing

a11.timing.Duration

Duration(nanoseconds: SupportsInt)

Creates a duration from a whole number of nanoseconds.

nanoseconds_value property

nanoseconds_value: int

The duration as a whole number of nanoseconds.

microseconds staticmethod

microseconds(value: SupportsFloat | None) -> Duration

Creates a duration from microseconds; None or negative means infinite.

milliseconds staticmethod

milliseconds(value: SupportsFloat | None) -> Duration

Creates a duration from milliseconds; None or negative means infinite.

nanoseconds staticmethod

nanoseconds(value: SupportsInt | None) -> Duration

Creates a duration from nanoseconds; None or negative means infinite.

seconds staticmethod

seconds(value: SupportsFloat | None) -> Duration

Creates a duration from seconds; None or negative means infinite.

float_seconds

float_seconds(infinity_value: SupportsFloat | None = None) -> float | None

Returns the duration in seconds as a float; infinity_value is returned for a positive-infinite duration.

is_infinite

is_infinite() -> bool

Returns whether the duration is positive or negative infinite.

a11.timing.Time

Time(nanoseconds_since_epoch: SupportsInt)

Creates a time from nanoseconds since the Unix epoch.

nanoseconds_since_epoch property

nanoseconds_since_epoch: int

The time as nanoseconds since the Unix epoch.

from_nanoseconds_since_epoch staticmethod

from_nanoseconds_since_epoch(nanoseconds: SupportsInt) -> Time

Creates a time from nanoseconds since the Unix epoch.