Benzene

Hosting on Google Cloud Functions (HTTP + Pub/Sub)

Host one set of Benzene handlers on Google Cloud Functions behind two triggers — an HTTP function and a Pub/Sub function — and publish events back out over Pub/Sub. The handlers are written once and never change between transports.

Prerequisites

1. The handlers (transport-agnostic)

Handlers that need a collaborator are built by a factory that closes over it — this is what makes them hostable anywhere and testable with a fake:

from benzene.core import Handler, MessageSender
from benzene.results import Result

def make_place_order(service, sender: MessageSender) -> Handler:
    async def place_order(request: PlaceOrder) -> Result:   # annotation → request_type is inferred
        order = service.place(request.sku, request.quantity)
        await sender.send_message("orders:created", {"id": order.id, "sku": order.sku})
        return Result.created(order)
    return place_order

2. Wire routes + topics

from benzene.core import Registry
from benzene.http import HttpRouter

router = HttpRouter().register("POST", "/orders", "orders:place", make_place_order(service, sender))

registry = Registry.from_definitions(router).register(   # the HTTP topics + the Pub/Sub subscriber
    "orders:created", make_on_order_created(...)
)

No request_type= is passed: place_order annotates its request (request: PlaceOrder), so Benzene infers the payload type from the signature and builds the body into it. This is the direct path — routes and topics wired by hand, no composition root.

3. Build the host and expose entry points

# main.py
from benzene.gcp import GcpFunctionsApp, http_function, pubsub_function

app = GcpFunctionsApp(http_router=router, registry=registry)
orders_http = http_function(app)
orders_pubsub = pubsub_function(app)

The HTTP trigger resolves the topic from the route and maps the Benzene status to an HTTP code; the Pub/Sub trigger reads the topic from the message's topic attribute and raises on failure so Pub/Sub redelivers.

4. Test it in memory (dogfooded, no cloud)

The direct-path host needs no composition root to test: wrap the same GcpFunctionsApp you built in step 3 in a GcpFunctionsTestHost and drive both triggers in memory, faking only the outbound edge.

from benzene.gcp import GcpFunctionsApp
from benzene.gcp.testing import GcpFunctionsTestHost
from benzene.testing import FakeMessageSender

sender = FakeMessageSender()                     # records egress instead of calling Pub/Sub
router = HttpRouter().register("POST", "/orders", "orders:place", make_place_order(service, sender))
registry = Registry.from_definitions(router).register(
    "orders:created", make_on_order_created(seen)
)
host = GcpFunctionsTestHost(GcpFunctionsApp(http_router=router, registry=registry))

response = host.send_http("POST", "/orders", body={"sku": "ABC", "quantity": 2})
assert response.status_code == 201
assert sender.last_topic == "orders:created"     # ingress -> handler -> egress

host.send_pubsub("orders:created", body={"id": "ord-1", "sku": "ABC"})   # exercise the subscriber

That is the whole direct path: build the app, wrap it in the test host, push native events. When you want the one-line, provider-agnostic harness instead — create_test_host(StartUp).build_gcp(), swap .build_gcp() for .build_aws() and the same test runs on another cloud — wire the handlers through a BenzeneStartUp composition root, as the runnable examples/gcp_orders does. Both are covered in Two ways to wire a service.

5. Deploy

gcloud functions deploy orders-http --gen2 --runtime python312 --source . \
  --entry-point orders_http --trigger-http \
  --set-env-vars BENZENE_PUBSUB_TOPIC=projects/<project>/topics/orders

gcloud functions deploy orders-pubsub --gen2 --runtime python312 --source . \
  --entry-point orders_pubsub --trigger-topic orders

See also