Skip to content

Microservices

natsio.micro.add_service 

add_service(
    nc: Client,
    config: ServiceConfig | None = None,
    /,
    **kwargs: Any,
) -> Service

Create and start a micro service.

Pass a ServiceConfig, or the config fields as keyword arguments (add_service(nc, name="calc", version="1.0.0")). Enables the $SRV monitoring responders immediately; register handlers with Service.add_endpoint() / Service.add_group().

natsio.micro.Service 

Service(nc: Client, config: ServiceConfig)

A running micro service instance (create via add_service()).

Answers $SRV monitoring for its lifetime and routes requests to registered endpoints. Use it as an async context manager, or await stopped to block until it stops.

id property 

id: str

This instance's unique id (a fresh NUID).

started property 

started: datetime

When this instance started (UTC).

stopped property 

stopped: Future[None]

A future resolved when the service stops (the 'done' surface).

__await__ 

__await__() -> Generator[None, None, Service]

await is optional and completes immediately (no I/O) — the same muscle-memory tolerance as natsio.Client.subscribe().

add_group 

add_group(
    name: str, *, queue_group: str | None = None
) -> Group

Create a top-level group with the given subject prefix.

add_endpoint 

add_endpoint(
    name: str,
    handler: Handler,
    *,
    subject: str | None = None,
    queue_group: str | None = None,
    metadata: dict[str, str] | None = None,
) -> Endpoint

Register a top-level endpoint (subject defaults to name).

info 

info() -> InfoResponse

The service's INFO as a typed object (same content as $SRV.INFO).

stats 

stats() -> StatsResponse

The service's STATS as a typed object (same content as $SRV.STATS).

reset 

reset() -> None

Reset all endpoint counters and restart the started clock.

stop async 

stop() -> None

Drain every subscription (endpoints + monitoring) and mark stopped.

Idempotent, and drain-friendly: in-flight requests finish before their subscription closes. Resolves stopped.

natsio.micro.Group 

Group(service: Service, prefix: str, queue_group: str)

A dotted subject prefix under which endpoints are registered.

Groups nest (add_group) and inherit their queue group from the parent unless overridden.

__await__ 

__await__() -> Generator[None, None, Group]

await is optional and completes immediately (no I/O) — the same muscle-memory tolerance as natsio.Client.subscribe().

add_group 

add_group(
    name: str, *, queue_group: str | None = None
) -> Group

Derive a nested group, prefixed by this group's prefix.

add_endpoint 

add_endpoint(
    name: str,
    handler: Handler,
    *,
    subject: str | None = None,
    queue_group: str | None = None,
    metadata: dict[str, str] | None = None,
) -> Endpoint

Register an endpoint whose subject is prefixed by this group.

natsio.micro.Endpoint 

Endpoint(
    name: str,
    subject: str,
    queue_group: str,
    handler: Handler,
    metadata: dict[str, str],
)

A registered request handler and its live statistics.

Counters are plain integers mutated on the subscription's own task (single asyncio thread, no lock needed); processing_time is in nanoseconds.

average_processing_time property 

average_processing_time: int

Mean handler time in nanoseconds (0 before the first request).

__await__ 

__await__() -> Generator[None, None, Endpoint]

await is optional and completes immediately (no I/O) — the same muscle-memory tolerance as natsio.Client.subscribe().

natsio.micro.Request 

Request(msg: Msg)

A single service request, exposing the payload and reply helpers.

A handler is async def handler(req: Request) -> None. It responds with respond() (or respond_error() for a failure) exactly once; returning without responding is allowed but leaves the caller waiting.

subject property 

subject: str

The subject the request arrived on.

reply property 

reply: str | None

The reply subject, or None when the caller expects no response.

headers property 

headers: Headers | None

The request headers, or None when the message carried none.

payload property 

payload: bytes

The request payload.

data property 

data: bytes

Alias for payload.

respond async 

respond(
    payload: bytes | str = b"",
    *,
    headers: HeadersInput | None = None,
) -> None

Send a successful reply. No-op semantics are the caller's concern: calling twice publishes twice to the (already-consumed) reply inbox.

respond_error async 

respond_error(
    code: str,
    description: str,
    data: bytes | str = b"",
    *,
    headers: HeadersInput | None = None,
) -> None

Send an error reply, setting the Nats-Service-Error and Nats-Service-Error-Code headers. Counts toward the endpoint's num_errors and becomes its last_error.

natsio.micro.ServiceConfig dataclass 

ServiceConfig(
    *,
    name: str,
    version: str,
    description: str = "",
    metadata: dict[str, str] | None = None,
    queue_group: str = DEFAULT_QUEUE_GROUP,
    stats_handler: StatsHandler | None = None,
    error_handler: ErrorHandler | None = None,
)

Configuration for a micro service.

name and version are required and validated loudly on construction; queue_group (default "q") is inherited by every endpoint unless a group or endpoint overrides it. metadata is immutable once the service is created (ADR-32).

natsio.micro.EndpointStats dataclass 

EndpointStats(
    *,
    extra: dict[str, Any] = dict(),
    name: str = "",
    subject: str = "",
    queue_group: str = "",
    num_requests: int = 0,
    num_errors: int = 0,
    last_error: str = "",
    processing_time: int = 0,
    average_processing_time: int = 0,
    data: Any = None,
)

Bases: JsonModel

One endpoint's counters, as reported by $SRV.STATS.

processing_time and average_processing_time are integer nanoseconds (matching the Go time.Duration wire form). data is the optional, handler-defined custom payload and is omitted when absent.

natsio.micro.ServiceError 

ServiceError(
    description: str = "",
    *,
    code: str = "500",
    subject: str = "",
    endpoint: str = "",
)

Bases: NATSError

A service-side error surfaced to the configured error_handler.

Carries the code/description that were (or would be) returned to the caller, plus the subject the request arrived on and the endpoint name it was routed to. The originating exception, when any, is chained as __cause__.

natsio.micro.ServiceConfigError 

ServiceConfigError(description: str = '')

Bases: ConfigError

A service, group or endpoint was configured with invalid values.