Microservices

pip install austial      # `austial.microservices` ships in austial-microservices, which `austial` installs

A microservice is an application driven by messages instead of HTTP requests. Handlers are decorated with a pattern:

from austial import Controller
from austial.microservices import EventPattern, MessagePattern, Payload
 
 
@Controller()
class MathController:
    @MessagePattern({"cmd": "sum"})                 # request-response: the return value is the reply
    def accumulate(self, numbers: list[int] = Payload()) -> int:
        return sum(numbers)
 
    @EventPattern("user.created")                   # fire-and-forget: nothing goes back
    async def on_user_created(self, user: dict = Payload()) -> None: ...

A pattern is a string or a dict ({"cmd": "sum"}; key order doesn’t matter). Handlers may live on controllers or any provider. A message pattern has exactly one handler (two fail at startup); an event pattern may have several. Handler arguments: Payload() / Payload("key") for the data, Ctx() for an RpcContext(pattern, kind); an unmarked first parameter is the payload.

Running one#

from austial import AustialFactory
from austial.microservices import MicroserviceOptions, Transport
 
ms = AustialFactory.create_microservice(AppModule, MicroserviceOptions(Transport.TCP, host="0.0.0.0", port=3001))
await ms.listen()        # init hooks, then start the transport
...
await ms.close()

ms is a standalone application context, so ms.get(Token), lifecycle hooks and async providers all work.

Calling one#

from austial.microservices import ClientOptions, ClientProxy, ClientsModule, Transport
 
@Module(imports=[ClientsModule.register([ClientOptions(name="MATH", transport=Transport.TCP, host="math", port=3001)])])
class AppModule: ...
 
 
@Injectable()
class Calculator:
    def __init__(self, math: ClientProxy = Inject("MATH")):
        self.math = math
 
    async def total(self, numbers):
        return await self.math.send({"cmd": "sum"}, numbers)          # waits for the reply
    async def announce(self, user):
        await self.math.emit("user.created", user)                    # does not wait

Clients connect when the application initialises and close when it shuts down. send(pattern, data, timeout=...) raises asyncio.TimeoutError if no reply arrives (default ClientOptions.timeout, 5 s) and RpcError (with .payload) if the handler failed. Raise RpcException("message") or RpcException({...}) in a handler for a structured error; any other exception reaches the caller as {"status": "error", "message": "TypeName: text"}.

Transports#

Transport	Use
`Transport.TCP`	JSON lines over a TCP connection; many calls in flight on one connection. Plain TCP: no TLS or authentication, so keep it on a private network.
`Transport.LOCAL`	In-process, addressed by `name`/`broker`. Tests, or splitting a monolith before you split the deployment.
`Transport.REDIS`	Redis Pub/Sub (`url="redis://..."`). Needs `austial-microservices[redis]`. Pub/Sub has no consumer groups: every replica receives every message, so run one replica per service or use `REDIS_STREAMS` for load balancing.
`Transport.REDIS_STREAMS`	Redis Streams: replicas sharing `group` form a consumer group (each message goes to one replica, acknowledged after the handler), and a request published while no replica is up is delivered when one starts. Streams are capped at `extra={"maxlen": 10000}` entries. `[redis]`.
`Transport.NATS`	NATS subjects; replicas sharing `group` form a queue group, so each request reaches one replica. `[nats]`.
`Transport.KAFKA`	Kafka topics (`url="host:port"`); replicas share a consumer group (`group`). Topics must already exist. `[kafka]`.
`Transport.MQTT`	MQTT 5 (`url="host:port"`); replicas sharing `group` use shared subscriptions. `[mqtt]`.
`Transport.RMQ`	RabbitMQ queues (`url="amqp://..."`); replicas consuming a queue share the load. `[rabbitmq]`.
`Transport.GRPC`	gRPC. With `extra={"proto_path": ...}` real protobuf services including streaming (see below), `[grpc-proto]`; without it, a JSON channel between Austial services (`austial.Microservice`, methods `Request` and `Emit`), `[grpc]`.
`Transport.CUSTOM`	Anything else: implement `ServerTransport` (`listen(handler)`, `close()`) and `ClientTransport` (`connect`, `request`, `emit`, `close`) and pass `custom_server=` / `custom_client=`.

Broker transports share one request/reply protocol (JSON {"id", "data", "reply_to"}; one reply topic per client). Options: url (broker address), group (consumer group / queue group), prefix (separates environments on one broker) and extra (keyword arguments for the broker client: credentials, TLS). Install the matching extra, for example pip install 'austial-microservices[nats]'.

app.connect_microservice(MicroserviceOptions(Transport.NATS, url="nats://broker:4222", group="billing"))
ClientsModule.register([ClientOptions("BILLING", Transport.NATS, url="nats://broker:4222")])

Custom wire format#

Broker transports encode messages as JSON. Pass codec= to both sides to change that: any object with encode(dict) -> bytes and decode(bytes) -> dict (msgpack, compression, encryption, a schema registry). JsonCodec is the default. The TCP and gRPC transports have their own framing and ignore it.

class MsgpackCodec:
    def encode(self, message): return msgpack.packb(message)
    def decode(self, raw): return msgpack.unpackb(raw)
 
MicroserviceOptions(Transport.NATS, url="nats://broker:4222", codec=MsgpackCodec())
ClientOptions("BILLING", Transport.NATS, url="nats://broker:4222", codec=MsgpackCodec())

gRPC with Protocol Buffers#

Without proto_path, Transport.GRPC is a schema-less JSON channel between Austial services. Give it a .proto file and it serves and calls real protobuf services, interoperable with any gRPC client or server (pip install 'austial-microservices[grpc-proto]'; the .proto is compiled at startup, no generated code to commit). Messages are plain dicts in your code.

service HeroService {
  rpc FindOne (HeroById) returns (Hero);
  rpc FindMany (Query) returns (stream Hero);
  rpc Sum (stream Number) returns (Number);
  rpc Chat (stream Line) returns (stream Line);
}

from austial.microservices import GrpcMethod, Payload
 
@Controller()
class HeroController:
    @GrpcMethod("HeroService")                     # method defaults to the PascalCase function name: FindOne
    def find_one(self, data: dict = Payload()):
        return {"id": data["id"], "name": "Ada"}
 
    @GrpcMethod("HeroService", "FindMany")         # server streaming: yield dicts (or return an iterable)
    async def many(self, data: dict = Payload()):
        for n in range(data["count"]):
            yield {"id": n}
 
    @GrpcMethod("HeroService")                     # client streaming: the payload is an async iterator of dicts
    async def sum(self, numbers=Payload()):
        return {"value": sum([n["value"] async for n in numbers])}
 
app.connect_microservice(MicroserviceOptions(Transport.GRPC, port=50051, extra={"proto_path": "hero.proto"}))
 
# a client
heroes = ClientProxy(ClientOptions("HEROES", Transport.GRPC, port=50051, extra={"proto_path": "hero.proto"}))
await heroes.connect()
service = heroes.get_service("HeroService")
hero = await service.find_one({"id": 1})                     # unary
async for hero in service.find_many({"count": 3}): ...       # server streaming
total = await service.sum([{"value": 1}, {"value": 2}])      # client streaming: any (async) iterable

extra also takes include (extra import directories for the .proto), services (server: which services to serve; default every service in the file) and server_credentials / channel_credentials for TLS. Guards, pipes and interceptors apply as on other transports. Errors map to gRPC status: RpcException({"message": "...", "code": 5}) sets the status code, Forbidden resource is PERMISSION_DENIED, a missing handler is UNIMPLEMENTED, anything else is UNKNOWN; on the client a failed call raises RpcError with status, message and code. The tests call the server with a stub generated by protoc and call a protoc-generated server from the client, so the wire format is the standard one.

Guards, pipes and interceptors#

They work on message handlers like on routes; guards see get_type() == "rpc" and read the message with context.switch_to_rpc().get_data(). A guard that returns False answers Forbidden resource.

Hybrid applications: HTTP and messages together#

app = AustialFactory.create(AppModule)
app.connect_microservice(MicroserviceOptions(Transport.TCP, port=3001))     # starts and stops with the app

The same handlers and providers serve both. Pass auto_start=False and call await app.start_all_microservices() to start the transport yourself. Under austial serve/uvicorn it starts during ASGI startup, after the lifecycle hooks.