RequestResolver

__init__(run_all, *, batch_key=None, delay=None, collect_while=None, pre_check=None) -> None

def __init__(
    self,
    run_all: Callable[[Sequence[Entry[Q_contra]]], Effect[None, E_co]],
    *,
    batch_key: Callable[[Entry[Q_contra]], Hashable] | None = None,
    delay: Effect[None] | None = None,
    collect_while: Callable[[Collection[Entry[Q_contra]]], bool] | None = None,
    pre_check: Callable[[Entry[Q_contra]], bool] | None = None,
) -> None:
    if delay is not None:
        require_effect(delay, "delay")
    self._init(
        lambda entries, _key: run_all(entries),
        batch_key=batch_key or _default_key,
        delay=_YieldNow() if delay is None else delay,
        collect_while=collect_while or _always,
        pre_check=pre_check,
    )

make_grouped(*, key, resolver) -> RequestResolver[Any, Any]

@classmethod
def make_grouped(
    cls,
    *,
    key: Callable[[Entry[Any]], Hashable],
    resolver: Callable[[Sequence[Entry[Any]], Any], Effect[None, Any]],
) -> RequestResolver[Any, Any]:
    """Batch requests by ``key``; ``resolver`` runs each batch with its
    key (the first entry's, among equal keys)."""
    return RequestResolver._of(resolver, batch_key=key)

from_function(f) -> RequestResolver[Any]

@classmethod
def from_function(cls, f: Callable[[Entry[Any]], object]) -> RequestResolver[Any]:
    """Answer each request with ``f(entry)``. (Pyright cannot check
    ``f``'s result against the request's success type.)"""

    def run_all(entries: Sequence[Entry[Any]]) -> Effect[None]:
        def complete() -> None:
            for entry in entries:
                entry._complete(Success(f(entry)))

        return _Sync(complete)

    return RequestResolver(run_all)

from_function_batched(f) -> RequestResolver[Any]

@classmethod
def from_function_batched(
    cls, f: Callable[[Sequence[Entry[Any]]], Iterable[object]]
) -> RequestResolver[Any]:
    """Answer a batch with ``f(entries)``, one value per entry in
    order. Entries left without a value die uncompleted."""

    def run_all(entries: Sequence[Entry[Any]]) -> Effect[None]:
        def complete() -> None:
            for entry, value in zip(entries, f(entries), strict=False):
                entry._complete(Success(value))

        return _Sync(complete)

    return RequestResolver(run_all)

from_effect(f) -> RequestResolver[Any, Any]

@classmethod
def from_effect(
    cls, f: Callable[[Entry[Any]], Effect[object, Any]]
) -> RequestResolver[Any, Any]:
    """Answer each request with ``f(entry)``, all at once, each on a
    fiber of its own with the services of the fiber that made it."""
    from pyfect.batching._from_effect import from_effect

    return from_effect(f)

from_effect_tagged(handlers) -> RequestResolver[Any, Any]

@classmethod
def from_effect_tagged(cls, handlers: TaggedHandlers[Any, Any]) -> RequestResolver[Any, Any]:
    """Split a batch by request class and answer each group with its
    handler, all at once: one value per entry, in order. A failing
    handler fails its group (Effect's ``fromEffectTagged``, keyed by
    class instead of ``_tag``)."""
    from pyfect.batching._from_effect import from_effect_tagged

    return from_effect_tagged(handlers)

never() -> RequestResolver[object]

@staticmethod
def never() -> RequestResolver[object]:
    """A resolver whose batches never finish."""
    from pyfect.effect import never

    return RequestResolver(lambda _entries: never)

set_delay_effect(delay) -> RequestResolver[Q_contra, E_co]

def set_delay_effect(self, delay: Effect[None]) -> RequestResolver[Q_contra, E_co]:
    """Collect each batch until ``delay`` ends."""
    require_effect(delay, "delay")
    return self._with(delay=delay)

set_delay(duration) -> RequestResolver[Q_contra, E_co]

def set_delay(self, duration: Duration) -> RequestResolver[Q_contra, E_co]:
    """Collect each batch for ``duration``."""
    return self._with(delay=sleep(duration))

batch_n(n) -> RequestResolver[Q_contra, E_co]

def batch_n(self, n: int) -> RequestResolver[Q_contra, E_co]:
    """Run a batch as soon as it holds ``n`` requests (at least one)."""
    size = max(1, n)

    def collect_while(entries: Collection[Entry[Any]]) -> bool:
        return len(entries) < size

    return self._with(collect_while=collect_while)

grouped(f) -> RequestResolver[Q, E_co]

def grouped[Q](
    self: RequestResolver[Q, E_co], f: Callable[[Entry[Q]], Hashable]
) -> RequestResolver[Q, E_co]:
    """Batch requests by ``f``; equal keys share a batch."""
    return self._with(batch_key=f)

around(before, after) -> RequestResolver[Q, E_co | E2]

def around[Q, B, E2 = Never](
    self: RequestResolver[Q, E_co],
    before: Callable[[Sequence[Entry[Q]]], Effect[B, E2]],
    after: Callable[[Sequence[Entry[Q]], B], Effect[object, E2]],
) -> RequestResolver[Q, E_co | E2]:
    """Run ``before`` ahead of each batch and ``after`` once it ends,
    however it ends (Effect's ``acquireUseRelease``)."""
    run_all = self._run_all

    def wrapped(entries: Sequence[Entry[Q]], key: Hashable) -> Effect[None, E_co | E2]:
        def release(b: B) -> Callable[[Any], Any]:
            fin = cast("Effect[object, Never]", _call_user(after, entries, b))
            return lambda ex: _run_finalizer(fin, ex)

        return uninterruptible_mask(
            lambda restore: before(entries).flat_map(
                lambda b: _MatchCause(
                    restore(_call_user(run_all, entries, key)),
                    lambda c: release(b)(Failure(c)),
                    lambda v: release(b)(Success(v)),
                )
            )
        )

    return self._with(run_all=wrapped)

race(that) -> RequestResolver[Q, E_co | E2]

def race[Q, E2 = Never](
    self: RequestResolver[Q, E_co], that: RequestResolver[Q, E2]
) -> RequestResolver[Q, E_co | E2]:
    """Run each batch with both resolvers; the first to finish wins
    and the other is interrupted. The result does not group."""
    mine, theirs = self._run_all, that._run_all
    return RequestResolver._of(
        lambda entries, key: _call_user(mine, entries, key).race(
            _call_user(theirs, entries, key)
        )
    )

with_span(name, options=None, **span_options) -> RequestResolver[Q, E_co]

def with_span[Q](
    self: RequestResolver[Q, E_co],
    name: str,
    options: Callable[[Sequence[Entry[Q]]], SpanOptions] | None = None,
    **span_options: Unpack[SpanOptions],
) -> RequestResolver[Q, E_co]:
    """Run each batch inside a span, with a ``batchSize`` attribute and
    links to the spans the requests were made in. ``options`` computes
    the span options from the batch, in place of the keywords."""
    from pyfect.tracer import ParentSpan, SpanLink

    run_all = self._run_all

    def build(entries: Sequence[Entry[Q]], key: Hashable) -> Effect[None, E_co]:
        opts = span_options if options is None else options(entries)
        links = list(opts.get("links", ()))
        seen: set[int] = set()
        for entry in entries:
            span = entry.context.unsafe_get(ParentSpan)
            if span is not None and id(span) not in seen:
                seen.add(id(span))
                links.append(SpanLink(span))
        attributes = {"batchSize": len(entries), **opts.get("attributes", {})}
        return run_all(entries, key).with_span(
            name, **{**opts, "links": links, "attributes": attributes}
        )

    def traced(entries: Sequence[Entry[Q]], key: Hashable) -> Effect[None, E_co]:
        return _call_user(build, entries, key)

    return self._with(run_all=traced)

with_cache(*, capacity, strategy='lru') -> Effect[RequestResolver[Q_contra, E_co]]

def with_cache(
    self, *, capacity: int, strategy: CacheStrategy = "lru"
) -> Effect[RequestResolver[Q_contra, E_co]]:
    """A resolver remembering the outcome of up to ``capacity``
    requests, by equality: a repeated request is answered without a
    batch, and one made while an equal one is pending waits for it.
    Interrupted outcomes are not kept. ``strategy`` picks what goes
    first: the least recently used (``"lru"``) or the oldest
    (``"fifo"``)."""
    from pyfect.batching._with_cache import with_cache

    return _Sync(lambda: with_cache(self, capacity, strategy))