Call a service before the write

A handler needs to price an order from a remote service, then persist the priced order. Doing both in one handler holds the database transaction open across the network call. A two-phase handler splits the work: prepare runs the call outside the transaction and apply writes inside it, so the connection is held only for the write. The concept is covered in Transactions → Two-phase handlers; this is the wiring.

The runnable version lives at examples/recipes/two_phase_pricing/ and runs on the in-memory mock store — no infrastructure needed.

Split the handler¶

Subclass TwoPhaseDocumentHandler: prepare does the external/compute work and returns a payload; apply writes it via the handler's write port. The base holds the read and write ports (not the execution context), so each phase declares what it needs:

@attrs.define(slots=True, kw_only=True, frozen=True)
class PriceAndCreate(
    TwoPhaseDocumentHandler[QuoteRequest, int, ReadOrder, CreateOrder]
):
    """Quote the price (outside the tx), then create the priced order (inside it)."""

    pricing: PricingService

    async def prepare(self, args: QuoteRequest) -> int:
        # OUTSIDE the transaction — no connection held across this call.
        return await self.pricing.quote(args.item)

    async def apply(self, args: QuoteRequest, payload: int) -> ReadOrder:
        # INSIDE the transaction — self.writer is the command port.
        return await self.writer.create(CreateOrder(item=args.item, price=payload))

prepare runs under the read-only flag — use the read port there, the write port only in apply. It runs exactly once per invocation: a retry or hedge re-runs only apply, with the payload prepare already produced.

Register it¶

.two_phase() marks the operation; the transaction route scopes apply's transaction (required — enforced at freeze). TwoPhaseDocumentBuilder resolves the read/write ports from the context and hands them to the handler:

PRICE_AND_CREATE = "orders.price_and_create"

# TwoPhaseDocumentBuilder resolves the read/write ports from the context and hands
# them to the handler — so the handler declares its ports, not the whole context.
# .two_phase() splits prepare/apply; the tx route scopes apply's transaction.
REGISTRY = (
    OperationRegistry(
        handlers={
            PRICE_AND_CREATE: TwoPhaseDocumentBuilder(
                spec=ORDER_SPEC,
                build=lambda reader, writer: PriceAndCreate(
                    reader=reader, writer=writer, pricing=PRICING
                ),
            )
        }
    )
    .bind(PRICE_AND_CREATE)
    .two_phase()
    .bind_tx()
    .set_route("mock")
    .finish(deep=True)
    .freeze()
)

Call it¶

prepare runs outside the transaction, the engine opens it, apply writes, and the row commits:

async def place_priced_order(ctx: ExecutionContext) -> ReadOrder:
    created = await REGISTRY.resolve(PRICE_AND_CREATE, ctx)(QuoteRequest(item="widget"))

    # The write committed and is readable afterwards.
    stored = await ctx.document.query(ORDER_SPEC).get(created.id)

    if (
        stored is None  # pyright: ignore[reportUnnecessaryComparison]
        or stored.price != created.price
    ):
        raise RuntimeError("expected the committed price to match the prepared one")

    return created

Notes¶

• When to reach for it. Lazy transaction acquisition already keeps the connection off the pool until your first query, so pure compute or an external call before the first query needs nothing special. Use a two-phase handler when the call must sit between a read and the write that depends on it.
• No read/write atomicity across phases. A read in prepare runs outside apply's transaction. If apply depends on it, re-validate on write — an optimistic-concurrency rev check — rather than assuming the read still holds.
• One transaction. Two-phase wraps a single transaction around apply. For multiple transactions with compensation between external steps, reach for a saga, not more phases.