`benzene.core`

The transport-neutral message-handling engine: handlers and the @message decorator, the registry, the middleware pipeline, the per-invocation DI container, and the BenzeneMessage envelope entry point. Distribution: benzene-core (depends on benzene-results).

pip install benzene-core

Handlers and `@message`

A handler is async def handle(request) -> Result. The @message decorator tags it with a topic:

from benzene.core import message
from benzene.results import Result

@message("order:create", request_type=OrderRequest, response_type=OrderCreated)
async def create_order(request: OrderRequest) -> Result:
    ...

Parameter	Meaning
`topic`	the topic id the handler serves
`version`	payload/handler version (default `""`, the unversioned handler)
`request_type`	dataclass/type to build from the decoded body before calling (optional)
`response_type`	declared response type, for descriptors/tooling (optional)

The decorator leaves the function an ordinary callable; registration is a separate, explicit step.

`Registry`

Maps a (topic, version) pair to at most one handler. Registering the same pair twice is a startup error (DuplicateHandlerError), not a runtime ambiguity (core-concepts §2, §9).

from benzene.core import Registry

registry = (
    Registry()
    .add(create_order)                          # a @message-tagged function
    .register("order:get", get_order)           # explicit, no decorator
    .register("order:get", get_order_v2, version="2")
)

registry.find("order:get")           # unversioned handler
registry.find("order:get", "2")      # exact version match
registry.find("order:get", "9")      # None — no fuzzy fallback

Middleware pipeline

Middleware is async def mw(context, next) -> None, run in registration order (first registered is outermost). A middleware that does not await next() short-circuits the pipeline. The pipeline runs exactly once per invocation; the message router (topic → handler) is ordinary middleware, registered last (core-concepts §4).

from benzene.core import Context, MiddlewarePipeline, message_router

async def timing(context: Context, next) -> None:
    # ... before ...
    await next()
    # ... after ...

pipeline = MiddlewarePipeline([timing]).use(message_router(registry))
await pipeline.handle(Context("order:create", {"sku": "ABC"}))

`Context`

Carries the resolved topic, version, the native request, lower-cased headers, the per-invocation scope, and a result slot the router fills in. Transport adapters may subclass it to add invocation-scoped facts.

Dependency injection

A minimal container with per-invocation scoping and overridable defaults: the framework registers its defaults with try_add*, and an application's own add* wins (core-concepts §8).

from benzene.core import Container, Lifetime

container = Container()
container.add_singleton(Clock, lambda scope: SystemClock())
container.try_add_scoped(UnitOfWork, lambda scope: UnitOfWork())   # only if absent

scope = container.create_scope()          # one per invocation
scope.get_service(Clock)

Lifetimes: Lifetime.SINGLETON, SCOPED, TRANSIENT. Keys are arbitrary tokens (typically a type or a str).

The DI container mirrors .NET's Benzene.Dependencies; it is folded into benzene.core rather than shipped separately (the C# split existed for assembly isolation, which Python does not need).

The `BenzeneMessage` envelope

BenzeneMessageApplication is the transport-neutral entry point. It decodes a request envelope {topic, headers, body}, runs the pipeline with the router last, and encodes a response envelope {statusCode, headers, body}. body is always a pre-serialized JSON string (wire-contracts §1).

from benzene.core import BenzeneMessageApplication

app = BenzeneMessageApplication(registry)     # optional: pipeline=, container=
response = await app.handle_async(
    {"topic": "order:create", "headers": {}, "body": '{"sku": "ABC"}'}
)
# {"statusCode": "created", "headers": {"content-type": "application/json"}, "body": "..."}

The version header is benzene-version (VERSION_HEADER). Helpers encode_response(result) and error_payload(result) produce the response envelope and the problem-details error body ({"status", "detail"}) respectively.

Exports

BenzeneMessageApplication, Container, Context, DuplicateHandlerError, Handler, HandlerDefinition, Lifetime, Middleware, MiddlewarePipeline, Next, Registry, Scope, VERSION_HEADER, definition_of, encode_response, error_payload, message, message_router, to_jsonable, to_request.

benzene.core

Handlers and @message

Registry