GitHub

batching

PreviousNext

API reference for batching.

Request batching: Request, RequestResolver and request.

A request describes one piece of data to fetch; a resolver fetches many at once. Requests made through the same resolver in the same scheduler tick are collected into one batch, so N concurrent lookups become one call (the N+1 query problem)::

@dataclass(frozen=True)
class GetUser(fx.Request[User, NotFound]):
    id: int

def fetch_users(entries: Sequence[Entry[GetUser]]) -> fx.Effect[None]:
    def complete() -> None:
        rows = db.users_by_id([e.request.id for e in entries])
        for e in entries:
            e.complete_unsafe(fx.Success(rows[e.request.id]))

    return fx.sync(complete)

users = fx.RequestResolver(fetch_users)
names = yield from fx.for_each(
    ids, lambda i: fx.request(GetUser(i), users), concurrency="unbounded"
)
  • Entry: one request in a batch, with the services of the fiber that made it; complete it with complete_unsafe (or the effects complete, complete_effect, succeed, fail, fail_cause).
  • RequestResolver(run_all, ...): Effect's make / makeWith; make_grouped, from_function, from_function_batched, from_effect, from_effect_tagged and never build others. Methods derive new resolvers: batch_n, grouped, set_delay, set_delay_effect, around, race and with_cache.
  • request(req, resolver): make a request from an effect; request_unsafe makes one from plain code with a callback.

Interrupting a caller takes its request out of its batch; a batch nobody waits for does not run.

Classes

UncompletedRequestErrorEntryRequestTypedRequestRequestResolver

Functions

request

request(req, resolver) -> Effect[Any, Any, Any]

Make req through resolver (or the resolver an effect builds) and wait for its outcome. Requests made in the same scheduler tick are batched::

name = yield from fx.request(GetName(user_id), names)

Interrupting the caller takes its request out of the batch; a batch nobody waits for any more does not run.

ParameterTypeDescription
reqTypedRequest[Any, Any, Any, Any]
resolverRequestResolver[Any, Any] | Effect[RequestResolver[Any, Any], Any, Any]

Returns Effect[Any, Any, Any]

def request(
    req: TypedRequest[Any, Any, Any, Any],
    resolver: RequestResolver[Any, Any] | Effect[RequestResolver[Any, Any], Any, Any],
    /,
) -> Effect[Any, Any, Any]:
    """Make ``req`` through ``resolver`` (or the resolver an effect builds)
    and wait for its outcome. Requests made in the same scheduler tick are
    batched::

        name = yield from fx.request(GetName(user_id), names)

    Interrupting the caller takes its request out of the batch; a batch
    nobody waits for any more does not run."""
    if isinstance(resolver, RequestResolver):
        return request_with(req, resolver)
    _require_resolver_effect(resolver)
    return resolver.flat_map(lambda r: request_with(req, r))

request_unsafe

request_unsafe(req, *, resolver, on_exit, context, scheduler=None) -> Callable[[], None]

Make req outside an effect: on_exit gets its outcome. The batch runs on the current fiber's run, the running asyncio loop, or scheduler (as run_fork). Returns a function that withdraws the request; on_exit is then never called.

ParameterTypeDescription
reqTypedRequest[A, E, Any, Q]
resolverRequestResolver[Q, E]
on_exitCallable[[Exit[A, E]], None]
contextContext
schedulerScheduler | None = None

Returns Callable[[], None]

def request_unsafe[A, E, Q](
    req: TypedRequest[A, E, Any, Q],
    *,
    resolver: RequestResolver[Q, E],
    on_exit: Callable[[Exit[A, E]], None],
    context: Context,
    scheduler: Scheduler | None = None,
) -> Callable[[], None]:
    """Make ``req`` outside an effect: ``on_exit`` gets its outcome. The
    batch runs on the current fiber's run, the running asyncio loop, or
    ``scheduler`` (as ``run_fork``). Returns a function that withdraws the
    request; ``on_exit`` is then never called."""
    run, now = ambient_run(scheduler)
    origin = Origin(context, run, None, now=now)
    entry = add_entry(resolver, req, on_exit, origin)
    return lambda: remove_entry(resolver, entry, origin)