Pull[A, E, L, R]: one step of a pull-based producer (Effect's Pull).
A pull is an effect that succeeds with the next item, fails with an ordinary
error E, or fails with Done(leftover) when the producer has finished
normally. Keeping the end in the error channel lets a consumer tell normal
completion from failure while every effect combinator still applies::
item = yield from pull # Done(...) ends the surrounding loopfilter_done and the helpers here separate Done from real failures:
catch_done and match_effect handle it, done_exit_from_cause
turns a cause into the producer's final Exit.
Pyright cannot subtract Done from an error union, so the functions that
take Done out of a type are data-first (catch_done(pull, f)): their
overloads read E | Done[L] off the pull itself.
Functions#
is_done_failure#
is_done_failure(reason) -> TypeIs[Fail[Done[Any]]]Whether reason is a Fail holding a Done.
| Parameter | Type | Description |
|---|---|---|
reason | Reason[object] |
Returns TypeIs[Fail[Done[Any]]]
def is_done_failure(reason: Reason[object]) -> TypeIs[Fail[Done[Any]]]:
"""Whether ``reason`` is a ``Fail`` holding a ``Done``."""
return _is_done(reason)is_done_cause#
is_done_cause(cause) -> boolWhether cause holds any Done failure.
| Parameter | Type | Description |
|---|---|---|
cause | Cause[object] |
Returns bool
def is_done_cause(cause: Cause[object]) -> bool:
"""Whether ``cause`` holds any ``Done`` failure."""
return any(_is_done(r) for r in cause.reasons)filter_no_done#
filter_no_done(cause) -> Result[Cause[E], Cause[E]]Ok(cause) when it holds no Done failure, else Err(cause).
| Parameter | Type | Description |
|---|---|---|
cause | Cause[E] |
Returns Result[Cause[E], Cause[E]]
def filter_no_done[E](cause: Cause[E]) -> Result[Cause[E], Cause[E]]:
"""``Ok(cause)`` when it holds no ``Done`` failure, else ``Err(cause)``."""
return Err(cause) if is_done_cause(cause) else Ok(cause)filter_done_leftover#
filter_done_leftover(cause) -> Result[Any, Cause[E]]filter_done, keeping only the Done's leftover value.
| Parameter | Type | Description |
|---|---|---|
cause | Cause[E] |
Returns Result[Any, Cause[E]]
def filter_done_leftover[E](cause: Cause[E]) -> Result[Any, Cause[E]]:
"""``filter_done``, keeping only the ``Done``'s leftover value."""
found = filter_done(cause)
return Ok(found.value.value) if isinstance(found, Ok) else founddone_exit_from_cause#
done_exit_from_cause(cause) -> Exit[Any, E]The producer's final exit: success with the leftover when Done
is the only failure, else a failure with the cause stripped of
Done.
| Parameter | Type | Description |
|---|---|---|
cause | Cause[E] |
Returns Exit[Any, E]
def done_exit_from_cause[E](cause: Cause[E]) -> Exit[Any, E]:
"""The producer's final exit: success with the leftover when ``Done``
is the only failure, else a failure with the cause stripped of
``Done``."""
found = filter_done(cause)
return Success(found.value.value) if isinstance(found, Ok) else Failure(found.error)catch_done#
catch_done(pull, f) -> Effect[Any, Any, Any]Recover from Done with f(leftover); other failures pass
through (a Done merged with a real failure is dropped from it).
| Parameter | Type | Description |
|---|---|---|
pull | Effect[Any, Any, Any] | |
f | Callable[[Any], Effect[Any, Any, Any]] |
Returns Effect[Any, Any, Any]
def catch_done(
pull: Effect[Any, Any, Any], f: Callable[[Any], Effect[Any, Any, Any]]
) -> Effect[Any, Any, Any]:
"""Recover from ``Done`` with ``f(leftover)``; other failures pass
through (a ``Done`` merged with a real failure is dropped from it)."""
require_effect(pull, "pull")
def on_failure(cause: Cause[Any]) -> Effect[Any, Any, Any]:
found = filter_done(cause)
return f(found.value.value) if isinstance(found, Ok) else _FailCause(found.error)
return _MatchCause(pull, on_failure, None)match_effect#
match_effect(pull, *, on_success, on_failure, on_done) -> Effect[Any, Any, Any]Handle each outcome of pull: an item, a failure (without
Done), or the end with its leftover.
| Parameter | Type | Description |
|---|---|---|
pull | Effect[Any, Any, Any] | |
on_success | Callable[[Any], Effect[Any, Any, Any]] | |
on_failure | Callable[[Cause[Any]], Effect[Any, Any, Any]] | |
on_done | Callable[[Any], Effect[Any, Any, Any]] |
Returns Effect[Any, Any, Any]
def match_effect(
pull: Effect[Any, Any, Any],
*,
on_success: Callable[[Any], Effect[Any, Any, Any]],
on_failure: Callable[[Cause[Any]], Effect[Any, Any, Any]],
on_done: Callable[[Any], Effect[Any, Any, Any]],
) -> Effect[Any, Any, Any]:
"""Handle each outcome of ``pull``: an item, a failure (without
``Done``), or the end with its leftover."""
require_effect(pull, "pull")
def failed(cause: Cause[Any]) -> Effect[Any, Any, Any]:
found = filter_done(cause)
return on_done(found.value.value) if isinstance(found, Ok) else on_failure(found.error)
return _MatchCause(pull, failed, on_success)