Sink

__init__(transform) -> None

def __init__(self, transform: Transform[A_co, In_contra, L_co, E_co, R_co]) -> None:
    self._build: Transform[Any, Any, Any, Any, Any] = transform

pipe(f) -> B

def pipe[B](self, f: Callable[[Self], B], /) -> B:
    """Apply ``f`` to this sink (one step; see ``Effect.pipe``)."""
    return f(self)

map(f) -> Sink[B, In_contra, L_co, E_co, R_co]

def map[B](self, f: Callable[[A_co], B]) -> Sink[B, In_contra, L_co, E_co, R_co]:
    """The sink with its value replaced by ``f(value)``."""
    from pyfect.sink import _combinators

    return _combinators.map_end(self, lambda end: (f(end[0]), end[1]))

as_(value) -> Sink[B, In_contra, L_co, E_co, R_co]

def as_[B](self, value: B) -> Sink[B, In_contra, L_co, E_co, R_co]:
    """The sink with its value replaced by ``value``."""
    from pyfect.sink import _combinators

    return _combinators.map_end(self, lambda end: (value, end[1]))

map_effect(f) -> Sink[B, In_contra, L_co, E_co | E2, R_co | R2]

def map_effect[B, E2 = Never, R2 = Never](
    self, f: Callable[[A_co], Effect[B, E2, R2]]
) -> Sink[B, In_contra, L_co, E_co | E2, R_co | R2]:
    """The sink with its value replaced by what ``f(value)`` succeeds
    with."""
    from pyfect.sink import _combinators

    return _combinators.map_effect_end(self, lambda end: f(end[0]).map(lambda b: (b, end[1])))

map_end(f) -> Sink[B, In_contra, L2, E_co, R_co]

def map_end[B, L2 = Never](
    self, f: Callable[[End[A_co, L_co]], End[B, L2]]
) -> Sink[B, In_contra, L2, E_co, R_co]:
    """The sink with its whole ``End`` (value and leftovers) replaced by
    ``f(end)``."""
    from pyfect.sink import _combinators

    return _combinators.map_end(self, f)

map_effect_end(f) -> Sink[B, In_contra, L2, E_co | E2, R_co | R2]

def map_effect_end[B, L2 = Never, E2 = Never, R2 = Never](
    self, f: Callable[[End[A_co, L_co]], Effect[End[B, L2], E2, R2]]
) -> Sink[B, In_contra, L2, E_co | E2, R_co | R2]:
    """``map_end`` with an effectful ``f``."""
    from pyfect.sink import _combinators

    return _combinators.map_effect_end(self, f)

summarized(summary, f) -> Sink[tuple[A_co, B], In_contra, L_co, E_co | E2, R_co | R2]

def summarized[S, B, E2 = Never, R2 = Never](
    self, summary: Effect[S, E2, R2], f: Callable[[S, S], B]
) -> Sink[tuple[A_co, B], In_contra, L_co, E_co | E2, R_co | R2]:
    """Run ``summary`` before and after the sink and pair its value with
    ``f(before, after)``."""
    from pyfect._effect_args import require_effect
    from pyfect.sink import _combinators

    require_effect(summary, "summary")
    return _combinators.summarized(self, summary, f)

with_duration() -> Sink[tuple[A_co, float], In_contra, L_co, E_co, R_co]

def with_duration(self) -> Sink[tuple[A_co, float], In_contra, L_co, E_co, R_co]:
    """Pair the value with how long the sink ran, in seconds, measured in
    whole monotonic nanoseconds."""
    from pyfect.sink import _combinators

    return _combinators.with_duration(self)

map_input(f) -> Sink[A_co, In0, L_co, E_co, R_co]

def map_input[In0](self, f: Callable[[In0], In_contra]) -> Sink[A_co, In0, L_co, E_co, R_co]:
    """A sink reading ``In0`` items, each given to this sink as
    ``f(item)``."""
    from pyfect.sink import _combinators

    return _combinators.map_input_array(self, lambda chunk: tuple(map(f, chunk)))

map_input_effect(f) -> Sink[A_co, In0, L_co, E_co | E2, R_co | R2]

def map_input_effect[In0, E2 = Never, R2 = Never](
    self, f: Callable[[In0], Effect[In_contra, E2, R2]]
) -> Sink[A_co, In0, L_co, E_co | E2, R_co | R2]:
    """``map_input`` with an effectful ``f``, run on each item in
    order."""
    from pyfect.sink import _combinators

    return _combinators.map_input_effect(self, f)

map_input_array(f) -> Sink[A_co, In0, L_co, E_co, R_co]

def map_input_array[In0](
    self, f: Callable[[tuple[In0, ...]], tuple[In_contra, ...]]
) -> Sink[A_co, In0, L_co, E_co, R_co]:
    """A sink reading ``In0`` chunks, each given to this sink as
    ``f(chunk)``, which must be non-empty."""
    from pyfect.sink import _combinators

    return _combinators.map_input_array(self, f)

map_input_array_effect(f) -> Sink[A_co, In0, L_co, E_co | E2, R_co | R2]

def map_input_array_effect[In0, E2 = Never, R2 = Never](
    self, f: Callable[[tuple[In0, ...]], Effect[tuple[In_contra, ...], E2, R2]]
) -> Sink[A_co, In0, L_co, E_co | E2, R_co | R2]:
    """``map_input_array`` with an effectful ``f``."""
    from pyfect.sink import _combinators

    return _combinators.map_input_array_effect(self, f)

map_leftover(f) -> Sink[A_co, In_contra, L2, E_co, R_co]

def map_leftover[L2](self, f: Callable[[L_co], L2]) -> Sink[A_co, In_contra, L2, E_co, R_co]:
    """The sink with each leftover item replaced by ``f(item)``."""
    from pyfect.sink import _combinators

    return _combinators.map_end(
        self, lambda end: (end[0], None if end[1] is None else tuple(map(f, end[1])))
    )

ignore_leftover() -> Sink[A_co, In_contra, Never, E_co, R_co]

def ignore_leftover(self) -> Sink[A_co, In_contra, Never, E_co, R_co]:
    """The sink without its leftovers: what it read but did not use is
    dropped."""
    from pyfect.sink import _combinators

    return _combinators.map_end(self, lambda end: (end[0], None))

flat_map(f) -> Sink[Any, Any, Any, Any, Any]

def flat_map(
    self, f: Callable[[A_co], Sink[Any, Any, Any, Any, Any]]
) -> Sink[Any, Any, Any, Any, Any]:
    """Run this sink, then ``f(value)`` on the rest of the input, its
    leftovers first. Leftovers the second sink does not read stay
    leftovers."""
    from pyfect.sink import _combinators

    return _combinators.flat_map(self, f)

map_error(f) -> Sink[A_co, In_contra, L_co, E2, R_co]

def map_error[E2](self, f: Callable[[E_co], E2]) -> Sink[A_co, In_contra, L_co, E2, R_co]:
    """The sink with its typed error replaced by ``f(error)``."""
    from pyfect.sink import _combinators

    return _combinators.transform_effect(self, lambda eff: eff.map_error(f))

catch_all(f) -> Sink[A_co | B, In_contra, L_co, E2, R_co | R2]

def catch_all[B, E2 = Never, R2 = Never](
    self, f: Callable[[E_co], Effect[B, E2, R2]]
) -> Sink[A_co | B, In_contra, L_co, E2, R_co | R2]:
    """Recover from a typed failure with the value of ``f(error)``
    (Effect's ``Sink.catch``). The stream's own failures pass by."""
    from pyfect.sink import _combinators

    return _combinators.catch_all(self, f)

catch_cause(f) -> Sink[A_co | B, In_contra, L_co, E2, R_co | R2]

def catch_cause[B, E2 = Never, R2 = Never](
    self, f: Callable[[Cause[E_co]], Effect[B, E2, R2]]
) -> Sink[A_co | B, In_contra, L_co, E2, R_co | R2]:
    """Recover from any failure of the sink with the value of
    ``f(cause)``. The stream's own failures pass by."""
    from pyfect.sink import _combinators

    return _combinators.catch_cause(self, f)

or_else(f) -> Sink[A_co | B, I, L_co | L2, E2, R_co | R2]

def or_else[I, B, L2 = Never, E2 = Never, R2 = Never](
    self: Sink[A_co, I, L_co, E_co, R_co], f: Callable[[E_co], Sink[B, I, L2, E2, R2]]
) -> Sink[A_co | B, I, L_co | L2, E2, R_co | R2]:
    """After a typed failure, go on with ``f(error)`` on the rest of
    the input. The stream's own failures pass by."""
    from pyfect.sink import _combinators

    return _combinators.or_else(self, f)

on_exit(f) -> Sink[A_co, In_contra, L_co, E_co, R_co | R2]

def on_exit[R2 = Never](
    self, f: Callable[[Exit[A_co, E_co]], Effect[object, Never, R2]]
) -> Sink[A_co, In_contra, L_co, E_co, R_co | R2]:
    """Run ``f(exit)`` when the sink ends. A stream failure stops the
    sink as an interruption would, so ``f`` sees an interrupt."""
    from pyfect.sink import _combinators

    return _combinators.on_exit(self, f)

ensuring(finalizer) -> Sink[A_co, In_contra, L_co, E_co, R_co | R2]

def ensuring[R2 = Never](
    self, finalizer: Effect[object, Never, R2]
) -> Sink[A_co, In_contra, L_co, E_co, R_co | R2]:
    """Run ``finalizer`` however the sink ends."""
    from pyfect._effect_args import require_effect
    from pyfect.sink import _combinators

    require_effect(finalizer, "finalizer")
    return _combinators.on_exit(self, lambda _: finalizer)