Channel

__init__(transform) -> None

def __init__(self, transform: Transform) -> None:
    self._build = transform

pipe(f) -> B

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

run_collect() -> Effect[list[O], E, R]

def run_collect[O, E, D, R](
    self: RunMethods[O, E, D, Never, Never, None, R],
) -> Effect[list[O], E, R]:
    """Run the channel and collect every item written."""
    from pyfect.channel import _run

    return _run.run_collect(self)

run_drain() -> Effect[D, E, R]

def run_drain[O, E, D, R](self: RunMethods[O, E, D, Never, Never, None, R]) -> Effect[D, E, R]:
    """Run the channel, dropping its items; succeed with its end value."""
    from pyfect.channel import _run

    return _run.run_drain(self)

run_for_each(f) -> Effect[D, E | E2, R | R2]

def run_for_each[O, E, D, R, E2 = Never, R2 = Never](
    self: RunMethods[O, E, D, Never, Never, None, R], f: Callable[[O], Effect[object, E2, R2]]
) -> Effect[D, E | E2, R | R2]:
    """Run ``f`` on each item; succeed with the channel's end value."""
    from pyfect.channel import _run

    return _run.run_for_each(self, f)

run_for_each_while(f) -> Effect[None, E | E2, R | R2]

def run_for_each_while[O, E, D, R, E2 = Never, R2 = Never](
    self: RunMethods[O, E, D, Never, Never, None, R], f: Callable[[O], Effect[bool, E2, R2]]
) -> Effect[None, E | E2, R | R2]:
    """Run ``f`` on each item until it succeeds with ``False``."""
    from pyfect.channel import _run

    return _run.run_for_each_while(self, f)

run_fold(initial, f) -> Effect[Z, E, R]

def run_fold[O, E, D, R, Z](
    self: RunMethods[O, E, D, Never, Never, None, R],
    initial: Callable[[], Z],
    f: Callable[[Z, O], Z],
) -> Effect[Z, E, R]:
    """Fold the items into ``initial()`` with ``f``."""
    from pyfect.channel import _run

    return _run.run_fold(self, initial, f)

run_fold_effect(initial, f) -> Effect[Z, E | E2, R | R2]

def run_fold_effect[O, E, D, R, Z, E2 = Never, R2 = Never](
    self: RunMethods[O, E, D, Never, Never, None, R],
    initial: Callable[[], Z],
    f: Callable[[Z, O], Effect[Z, E2, R2]],
) -> Effect[Z, E | E2, R | R2]:
    """Fold the items into ``initial()`` with the effect ``f``."""
    from pyfect.channel import _run

    return _run.run_fold_effect(self, initial, f)

run_count() -> Effect[int, E, R]

def run_count[O, E, D, R](
    self: RunMethods[O, E, D, Never, Never, None, R],
) -> Effect[int, E, R]:
    """Run the channel and count its items."""
    from pyfect.channel import _run

    return _run.run_fold(self, _zero, _increment)

run_head() -> Effect[Option[O], E, R]

def run_head[O, E, D, R](
    self: RunMethods[O, E, D, Never, Never, None, R],
) -> Effect[Option[O], E, R]:
    """The first item (``Some``), or ``None`` when there is none. Stops
    the channel after the first item."""
    from pyfect.channel import _run

    return _run.run_head(self)

run_last() -> Effect[Option[O], E, R]

def run_last[O, E, D, R](
    self: RunMethods[O, E, D, Never, Never, None, R],
) -> Effect[Option[O], E, R]:
    """The last item (``Some``), or ``None`` when there is none."""
    from pyfect.channel import _run

    return _run.run_last(self)

run_bytes() -> Effect[bytes, E, R]

def run_bytes[E, D, R](
    self: RunMethods[tuple[bytes, ...], E, D, Never, Never, None, R],
) -> Effect[bytes, E, R]:
    """Run a channel of byte chunks and join them (Effect's
    ``mkUint8Array``)."""
    from pyfect.channel import _run

    return _run.run_fold(self, list[bytes], _extend).map(b"".join)

to_pull() -> Effect[Pull[O, E, D], Never, R | Scope]

def to_pull[O, E, D, R](
    self: RunMethods[O, E, D, Never, Never, None, R],
) -> Effect[Pull[O, E, D], Never, R | Scope]:
    """Build the channel's pull in the current ``Scope``, which holds
    its resources. The pull has the caller's services and takes turns:
    fibers pulling at once get one item each."""
    from pyfect.channel import _run

    return _run.to_pull(self)

to_pull_scoped(scope) -> Effect[Pull[O, E, D, R], Never, R]

def to_pull_scoped[O, E, D, R](
    self: RunMethods[O, E, D, Never, Never, None, R], scope: CloseableScope
) -> Effect[Pull[O, E, D, R], Never, R]:
    """Build the channel's pull with its resources in ``scope``."""
    from pyfect.channel import _run

    return _run.to_pull_scoped(self, scope)

run_into_queue(queue) -> Effect[None, Never, R]

def run_into_queue[O, E, D, R](
    self: RunMethods[O, E, D, Never, Never, None, R], queue: Enqueue[O, E | Done]
) -> Effect[None, Never, R]:
    """Offer every item to ``queue``, then end it, or fail it with the
    channel's cause. The queue is finished even if the run is
    interrupted."""
    from pyfect.channel import _sinks

    return _sinks.run_into_queue(self, queue, array=False)

run_into_queue_array(queue) -> Effect[None, Never, R]

def run_into_queue_array[A, E, D, R](
    self: RunMethods[tuple[A, ...], E, D, Never, Never, None, R], queue: Enqueue[A, E | Done]
) -> Effect[None, Never, R]:
    """``run_into_queue``, offering the items of each tuple."""
    from pyfect.channel import _sinks

    return _sinks.run_into_queue(self, queue, array=True)

to_queue(*, capacity, strategy='suspend') -> Effect[Dequeue[O, E | Done], Never, R | Scope]

def to_queue[O, E, D, R](
    self: RunMethods[O, E, D, Never, Never, None, R],
    *,
    capacity: Capacity,
    strategy: Strategy = "suspend",
) -> Effect[Dequeue[O, E | Done], Never, R | Scope]:
    """Run the channel on a fiber in the current ``Scope``, offering its
    items to a new queue of ``capacity``; the queue ends (or fails) as
    the channel does and shuts down with the scope."""
    from pyfect.channel import _sinks

    return _sinks.to_queue(self, capacity, strategy, array=False)

to_queue_array(*, capacity, strategy='suspend') -> Effect[Dequeue[A, E | Done], Never, R | Scope]

def to_queue_array[A, E, D, R](
    self: RunMethods[tuple[A, ...], E, D, Never, Never, None, R],
    *,
    capacity: Capacity,
    strategy: Strategy = "suspend",
) -> Effect[Dequeue[A, E | Done], Never, R | Scope]:
    """``to_queue``, offering the items of each tuple."""
    from pyfect.channel import _sinks

    return _sinks.to_queue(self, capacity, strategy, array=True)

run_into_pubsub(pubsub, *, shutdown_on_end=False) -> Effect[D, E, R]

def run_into_pubsub[O, E, D, R](
    self: RunMethods[O, E, D, Never, Never, None, R],
    pubsub: PubSub[O],
    *,
    shutdown_on_end: bool = False,
) -> Effect[D, E, R]:
    """Publish every item to ``pubsub``; with ``shutdown_on_end``, shut it
    down when the run ends."""
    from pyfect.channel import _sinks

    return _sinks.run_into_pubsub(self, pubsub, shutdown_on_end, array=False)

run_into_pubsub_array(pubsub, *, shutdown_on_end=False) -> Effect[D, E, R]

def run_into_pubsub_array[A, E, D, R](
    self: RunMethods[tuple[A, ...], E, D, Never, Never, None, R],
    pubsub: PubSub[A],
    *,
    shutdown_on_end: bool = False,
) -> Effect[D, E, R]:
    """``run_into_pubsub``, publishing the items of each tuple."""
    from pyfect.channel import _sinks

    return _sinks.run_into_pubsub(self, pubsub, shutdown_on_end, array=True)

to_pubsub(*, capacity, strategy='suspend', replay=None, shutdown_on_end=True) -> Effect[PubSub[O], Never, R | Scope]

def to_pubsub[O, E, D, R](
    self: RunMethods[O, E, D, Never, Never, None, R],
    *,
    capacity: Capacity,
    strategy: Strategy = "suspend",
    replay: int | None = None,
    shutdown_on_end: bool = True,
) -> Effect[PubSub[O], Never, R | Scope]:
    """Run the channel on a fiber in the current ``Scope``, publishing
    its items to a new pubsub, shut down with the scope (and, unless
    ``shutdown_on_end`` is off, when the run ends)."""
    from pyfect.channel import _sinks

    return _sinks.to_pubsub(self, capacity, strategy, replay, shutdown_on_end, array=False)

to_pubsub_array(*, capacity, strategy='suspend', replay=None, shutdown_on_end=True) -> Effect[PubSub[A], Never, R | Scope]

def to_pubsub_array[A, E, D, R](
    self: RunMethods[tuple[A, ...], E, D, Never, Never, None, R],
    *,
    capacity: Capacity,
    strategy: Strategy = "suspend",
    replay: int | None = None,
    shutdown_on_end: bool = True,
) -> Effect[PubSub[A], Never, R | Scope]:
    """``to_pubsub``, publishing the items of each tuple."""
    from pyfect.channel import _sinks

    return _sinks.to_pubsub(self, capacity, strategy, replay, shutdown_on_end, array=True)

to_pubsub_take(*, capacity, strategy='suspend', replay=None) -> Effect[PubSub[Take[A, E, D]], Never, R | Scope]

def to_pubsub_take[A, E, D, R](
    self: RunMethods[tuple[A, ...], E, D, Never, Never, None, R],
    *,
    capacity: Capacity,
    strategy: Strategy = "suspend",
    replay: int | None = None,
) -> Effect[PubSub[Take[A, E, D]], Never, R | Scope]:
    """Run the channel on a fiber in the current ``Scope``, publishing
    each tuple as a ``Take``, then the run's exit as the final one."""
    from pyfect.channel import _sinks

    return _sinks.to_pubsub_take(self, capacity, strategy, replay)

on_start(eff) -> Channel[OutElem_co, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def on_start[E2 = Never, R2 = Never](
    self, eff: Effect[object, E2, R2]
) -> Channel[
    OutElem_co,
    OutErr_co | E2,
    OutDone_co,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """Run ``eff`` on the first pull, before the channel starts."""
    from pyfect._effect_args import require_effect
    from pyfect.channel import _lifecycle

    require_effect(eff, "eff")
    return _lifecycle.on_start(self, eff)

on_first(f) -> Channel[OutElem_co, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def on_first[E2 = Never, R2 = Never](
    self, f: Callable[[OutElem_co], Effect[object, E2, R2]]
) -> Channel[
    OutElem_co,
    OutErr_co | E2,
    OutDone_co,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """Run ``f`` on the first item, before writing it."""
    from pyfect.channel import _lifecycle

    return _lifecycle.on_first(self, f)

on_end(eff) -> Channel[OutElem_co, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def on_end[E2 = Never, R2 = Never](
    self, eff: Effect[object, E2, R2]
) -> Channel[
    OutElem_co,
    OutErr_co | E2,
    OutDone_co,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """Run ``eff`` when the channel ends (not when it fails)."""
    from pyfect._effect_args import require_effect
    from pyfect.channel import _lifecycle

    require_effect(eff, "eff")
    return _lifecycle.on_end(self, eff)

on_exit(f) -> Channel[OutElem_co, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def on_exit[R2 = Never](
    self, f: Callable[[Exit[OutDone_co, OutErr_co]], Effect[object, Never, R2]]
) -> Channel[
    OutElem_co, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2
]:
    """Run ``f(exit)`` as soon as the channel ends or fails, or when its
    run stops early."""
    from pyfect.channel import _lifecycle

    return _lifecycle.on_exit(self, f)

on_error(f) -> Channel[OutElem_co, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def on_error[R2 = Never](
    self, f: Callable[[Cause[OutErr_co]], Effect[object, Never, R2]]
) -> Channel[
    OutElem_co, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2
]:
    """``on_exit`` for a failure (or an early stop that interrupts)."""
    from pyfect.channel import _lifecycle

    return _lifecycle.on_exit(self, lambda ex: ex.match(on_success=_void, on_failure=f))

ensuring(finalizer) -> Channel[OutElem_co, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def ensuring[R2 = Never](
    self, finalizer: Effect[object, Never, R2]
) -> Channel[
    OutElem_co, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2
]:
    """Run ``finalizer`` however the channel stops."""
    from pyfect._effect_args import require_effect
    from pyfect.channel import _lifecycle

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

with_span(name, **options) -> Channel[OutElem_co, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co]

def with_span(
    self, name: str, **options: Unpack[SpanOptions]
) -> Channel[
    OutElem_co, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co
]:
    """Run inside a new span, the parent of the spans made inside. It
    ends with the channel's exit."""
    from pyfect.channel import _lifecycle

    return _lifecycle.with_span(self, name, options)

update_service(tag, f) -> Channel[OutElem_co, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | T]

def update_service[S, T](
    self, tag: TagLike[S, T], f: Callable[[S], S]
) -> Channel[
    OutElem_co, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | T
]:
    """Run with ``f`` applied to the service of ``tag``, which this
    channel then needs."""
    from pyfect.channel import _context

    return _context.update_context_with(self, lambda ctx: ctx.add(tag, f(ctx.unsafe_get(tag))))

repeat(schedule) -> Channel[Any, Any, Any, Any, Any, Any, Any]

def repeat(
    self, schedule: Schedule[Any, Any, Any, Any] | Callable[[Any], Any], /
) -> Channel[Any, Any, Any, Any, Any, Any, Any]:
    """Run the channel again after each end, stepping ``schedule`` with
    its end value, until the schedule stops; then end with the last
    end value. Each run sees its step's ``schedule.CurrentMetadata``."""
    from pyfect.channel import _repeat

    return _repeat.repeat(self, schedule)

forever() -> Channel[OutElem_co, OutErr_co, Never, InElem_contra, InErr_contra, InDone_contra, Env_co]

def forever(
    self,
) -> Channel[OutElem_co, OutErr_co, Never, InElem_contra, InErr_contra, InDone_contra, Env_co]:
    """Run the channel again each time it ends, until it fails."""
    from pyfect.channel import _repeat

    return _repeat.forever(self)

schedule(schedule) -> Channel[OutElem_co, OutErr_co | EE, OutDone_co | SO, InElem_contra, InErr_contra, InDone_contra, Env_co | RR]

def schedule[SO, EE = Never, RR = Never](
    self, schedule: Schedule[SO, OutElem_co, EE, RR]
) -> Channel[
    OutElem_co,
    OutErr_co | EE,
    OutDone_co | SO,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | RR,
]:
    """Step ``schedule`` with each item, writing it after the step's
    delay; when the schedule stops, end with its output (dropping the
    item that stopped it)."""
    from pyfect.channel import _repeat

    return _repeat.schedule(self, schedule)

merge(that, *, halt_strategy='both') -> Channel[OutElem_co | O2, OutErr_co | E2, OutDone_co | D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def merge[O2, E2 = Never, D2 = None, R2 = Never](
    self,
    that: Channel[O2, E2, D2, InElem_contra, InErr_contra, InDone_contra, R2],
    *,
    halt_strategy: HaltStrategy = "both",
) -> Channel[
    OutElem_co | O2,
    OutErr_co | E2,
    OutDone_co | D2,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """Run both channels at once, writing items as either writes them.
    The merge ends when both have ended, or just the ``"left"``
    (``self``) or ``"right"`` one, or ``"either"``; a failure of either
    fails it at once."""
    from pyfect.channel import _merge

    require_channel(that, "that")
    return _merge.merge(self, that, halt_strategy)

merge_effect(eff) -> Channel[OutElem_co, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def merge_effect[E2 = Never, R2 = Never](
    self, eff: Effect[object, E2, R2]
) -> Channel[
    OutElem_co,
    OutErr_co | E2,
    OutDone_co,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """Run ``eff`` alongside the channel: its failure fails the channel;
    it is interrupted when the channel ends."""
    from pyfect.channel import _merge

    return _merge.merge_effect(self, eff)

merge_all(*, concurrency, buffer_size=16, switch=False) -> Channel[O, EE, D, I, IE, ID, RR]

def merge_all[O, D1, D, I, IE, ID, EE = Never, RR = Never](
    self: ConcurrencyMethods[ChannelBase[O, EE, D1, I, IE, ID, RR], EE, D, I, IE, ID, RR],
    *,
    concurrency: Concurrency,
    buffer_size: int = 16,
    switch: bool = False,
) -> Channel[O, EE, D, I, IE, ID, RR]:
    """Run the channels this channel writes, up to ``concurrency`` at
    once, writing their items as they come (at most ``buffer_size``
    waiting). With ``switch``, a new channel past the limit interrupts
    the oldest one instead of waiting for it."""
    from pyfect.channel import _merge

    return _merge.merge_all(self, concurrency, buffer_size, switch=switch)

switch_map(f, *, concurrency=1, buffer_size=16) -> Channel[O2, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def switch_map[O2, E2 = Never, D2 = None, R2 = Never](
    self,
    f: Callable[
        [OutElem_co], Channel[O2, E2, D2, InElem_contra, InErr_contra, InDone_contra, R2]
    ],
    *,
    concurrency: Concurrency = 1,
    buffer_size: int = 16,
) -> Channel[
    O2, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2
]:
    """``flat_map`` where each new item interrupts the oldest running
    channel once ``concurrency`` are running."""
    from pyfect.channel import _combinators, _merge

    return _merge.merge_all(_combinators.map_(self, f), concurrency, buffer_size, switch=True)

buffer(*, capacity, strategy='suspend') -> Channel[OutElem_co, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co]

def buffer(
    self, *, capacity: int | Literal["unbounded"], strategy: Strategy = "suspend"
) -> Channel[
    OutElem_co, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co
]:
    """Pull ahead on a fiber, keeping up to ``capacity`` items for the
    reader; a full buffer waits, or drops the new (``"dropping"``) or
    oldest (``"sliding"``) item."""
    from pyfect.channel import _merge

    return _merge.buffer(self, capacity, strategy, array=False)

buffer_array(*, capacity, strategy='suspend') -> Channel[tuple[A, ...], E, D, I, IE, ID, R]

def buffer_array[A, E, D, I, IE, ID, R](
    self: ConcurrencyMethods[tuple[A, ...], E, D, I, IE, ID, R],
    *,
    capacity: int | Literal["unbounded"],
    strategy: Strategy = "suspend",
) -> Channel[tuple[A, ...], E, D, I, IE, ID, R]:
    """``buffer`` counting the items of each tuple; the reader gets all
    the buffered items at once."""
    from pyfect.channel import _merge

    return _merge.buffer(self, capacity, strategy, array=True)

interrupt_when(eff) -> Channel[OutElem_co, OutErr_co | E2, OutDone_co | D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def interrupt_when[D2, E2 = Never, R2 = Never](
    self, eff: Effect[D2, E2, R2]
) -> Channel[
    OutElem_co,
    OutErr_co | E2,
    OutDone_co | D2,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """Run ``eff`` alongside the channel; when it succeeds, interrupt the
    channel and end with its value; when it fails, fail."""
    from pyfect._effect_args import require_effect
    from pyfect.channel import _merge

    require_effect(eff, "eff")
    return _merge.interrupt_when(self, eff)

halt_when(eff) -> Channel[OutElem_co, OutErr_co | E2, OutDone_co | D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def halt_when[D2, E2 = Never, R2 = Never](
    self, eff: Effect[D2, E2, R2]
) -> Channel[
    OutElem_co,
    OutErr_co | E2,
    OutDone_co | D2,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """Run ``eff`` alongside the channel; once it has finished, the next
    pull ends with its value (or fails with its failure) instead of
    pulling."""
    from pyfect._effect_args import require_effect
    from pyfect.channel import _merge

    require_effect(eff, "eff")
    return _merge.halt_when(self, eff)

embed_input(input) -> Channel[Any, Any, Any, Any, Any, Any, Any]

def embed_input(
    self, input: Callable[[Any], Effect[object, Never, Any]]
) -> Channel[Any, Any, Any, Any, Any, Any, Any]:
    """A channel that hands its upstream to ``input``, run on a fiber,
    and runs this channel with no upstream."""
    from pyfect.channel import _merge

    return _merge.embed_input(self, input)

catch_cause(f) -> Channel[OutElem_co | O2, E2, OutDone_co | D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def catch_cause[O2, E2 = Never, D2 = None, R2 = Never](
    self,
    f: Callable[
        [Cause[OutErr_co]], Channel[O2, E2, D2, InElem_contra, InErr_contra, InDone_contra, R2]
    ],
) -> Channel[
    OutElem_co | O2,
    E2,
    OutDone_co | D2,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """On any failure, defects included, release this channel's
    resources and go on with ``f(cause)`` on the same upstream."""
    from pyfect.channel import _errors

    return _errors.catch_cause(self, f)

catch_cause_if(predicate, f) -> Channel[OutElem_co | O2, OutErr_co | E2, OutDone_co | D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def catch_cause_if[O2, E2 = Never, D2 = None, R2 = Never](
    self,
    predicate: Callable[[Cause[OutErr_co]], bool],
    f: Callable[
        [Cause[OutErr_co]], Channel[O2, E2, D2, InElem_contra, InErr_contra, InDone_contra, R2]
    ],
) -> Channel[
    OutElem_co | O2,
    OutErr_co | E2,
    OutDone_co | D2,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """``catch_cause`` for the causes that pass ``predicate``."""
    from pyfect.channel import _errors

    return _errors.catch_cause_if(self, predicate, f)

catch_all(f) -> Channel[OutElem_co | O2, E2, OutDone_co | D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def catch_all[O2, E2 = Never, D2 = None, R2 = Never](
    self,
    f: Callable[
        [OutErr_co], Channel[O2, E2, D2, InElem_contra, InErr_contra, InDone_contra, R2]
    ],
) -> Channel[
    OutElem_co | O2,
    E2,
    OutDone_co | D2,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """Go on with ``f(error)`` after a typed failure (Effect's
    ``Channel.catch``). Defects and interruptions pass through."""
    from pyfect.channel import _errors

    return _errors.catch_all(self, f)

catch_if(predicate, f, *, or_else=None) -> Channel[Any, Any, Any, Any, Any, Any, Any]

def catch_if(
    self,
    predicate: Callable[[Any], bool],
    f: Callable[[Any], Channel[Any, Any, Any, Any, Any, Any, Any]],
    *,
    or_else: Callable[[Any], Channel[Any, Any, Any, Any, Any, Any, Any]] | None = None,
) -> Channel[Any, Any, Any, Any, Any, Any, Any]:
    """Go on with ``f(error)`` after a typed failure that passes
    ``predicate``; ``or_else``, if given, handles the rest. To remove the
    caught errors from ``OutErr``, use ``pipe(channel.catch_if(...))``
    with a ``TypeIs`` refinement."""
    from pyfect.channel import _errors

    return _errors.catch_if(self, predicate, f, or_else)

catch_filter(filter, f, *, or_else=None) -> Channel[Any, Any, Any, Any, Any, Any, Any]

def catch_filter(
    self,
    filter: Callable[[Any], Result[Any, Any]],
    f: Callable[[Any], Channel[Any, Any, Any, Any, Any, Any, Any]],
    *,
    or_else: Callable[[Any], Channel[Any, Any, Any, Any, Any, Any, Any]] | None = None,
) -> Channel[Any, Any, Any, Any, Any, Any, Any]:
    """Go on with ``f(value)`` after a typed failure ``filter`` maps to
    ``Ok(value)``. One it maps to ``Err(x)`` goes to ``or_else(x)``, or
    fails with ``x``. Defects and interruptions pass through."""
    from pyfect.channel import _errors

    return _errors.catch_filter(self, filter, f, or_else)

catch_cause_filter(filter, f) -> Channel[OutElem_co | O2, X | E2, OutDone_co | D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def catch_cause_filter[EB, X, O2, E2 = Never, D2 = None, R2 = Never](
    self,
    filter: Callable[[Cause[OutErr_co]], Result[EB, Cause[X]]],
    f: Callable[
        [EB, Cause[OutErr_co]],
        Channel[O2, E2, D2, InElem_contra, InErr_contra, InDone_contra, R2],
    ],
) -> Channel[
    OutElem_co | O2,
    X | E2,
    OutDone_co | D2,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """Go on with ``f(value, cause)`` after a failure ``filter`` maps to
    ``Ok(value)``; one it maps to ``Err(rest)`` fails with ``rest``."""
    from pyfect.channel import _errors

    return _errors.catch_cause_filter(self, filter, f)

catch_defect(f) -> Channel[OutElem_co | O2, OutErr_co | E2, OutDone_co | D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def catch_defect[O2, E2 = Never, D2 = None, R2 = Never](
    self,
    f: Callable[
        [BaseException], Channel[O2, E2, D2, InElem_contra, InErr_contra, InDone_contra, R2]
    ],
) -> Channel[
    OutElem_co | O2,
    OutErr_co | E2,
    OutDone_co | D2,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """Go on with ``f(defect)`` after a failure with a defect (the
    first). Typed errors pass through."""
    from pyfect.channel import _errors

    return _errors.catch_defect(self, f)

tap_cause(f) -> Channel[OutElem_co, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def tap_cause[E2 = Never, R2 = Never](
    self, f: Callable[[Cause[OutErr_co]], Effect[object, E2, R2]]
) -> Channel[
    OutElem_co,
    OutErr_co | E2,
    OutDone_co,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """Run ``f(cause)`` when the channel fails, then fail the same way."""
    from pyfect.channel import _errors

    return _errors.tap_cause(self, f)

tap_error(f) -> Channel[OutElem_co, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def tap_error[E2 = Never, R2 = Never](
    self, f: Callable[[OutErr_co], Effect[object, E2, R2]]
) -> Channel[
    OutElem_co,
    OutErr_co | E2,
    OutDone_co,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """Run ``f(error)`` when the channel fails with a typed error."""
    from pyfect.channel import _errors

    return _errors.tap_error(self, f)

map_error(f) -> Channel[OutElem_co, E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co]

def map_error[E2](
    self, f: Callable[[OutErr_co], E2]
) -> Channel[OutElem_co, E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co]:
    """Apply ``f`` to each typed error. Defects and interruptions stay."""
    from pyfect.channel import _errors

    return _errors.map_error(self, f)

or_die() -> Channel[OutElem_co, Never, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co]

def or_die(
    self,
) -> Channel[OutElem_co, Never, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co]:
    """Die with the typed error instead of failing with it."""
    from pyfect.channel import _errors

    return _errors.or_die(self)

ignore(*, log=False) -> Channel[OutElem_co, Never, OutDone_co | None, InElem_contra, InErr_contra, InDone_contra, Env_co]

def ignore(
    self, *, log: bool | Severity = False
) -> Channel[
    OutElem_co, Never, OutDone_co | None, InElem_contra, InErr_contra, InDone_contra, Env_co
]:
    """End (with ``None``) instead of failing with a typed error. With
    ``log``, log the cause first, at the current log level (``True``)
    or the level given."""
    from pyfect.channel import _errors

    return _errors.ignore(self, log)

ignore_cause(*, log=False) -> Channel[OutElem_co, Never, OutDone_co | None, InElem_contra, InErr_contra, InDone_contra, Env_co]

def ignore_cause(
    self, *, log: bool | Severity = False
) -> Channel[
    OutElem_co, Never, OutDone_co | None, InElem_contra, InErr_contra, InDone_contra, Env_co
]:
    """End (with ``None``) instead of failing in any way. ``log`` as for
    ``ignore``."""
    from pyfect.channel import _errors

    return _errors.ignore_cause(self, log)

retry(policy) -> Channel[Any, Any, Any, Any, Any, Any, Any]

def retry(
    self, policy: Schedule[Any, Any, Any, Any] | Callable[[Any], Any], /
) -> Channel[Any, Any, Any, Any, Any, Any, Any]:
    """Run the channel again from the start after each typed failure,
    while ``policy`` recurs; then fail with the last error. Items
    already written stay written. A run that writes an item resets
    the policy."""
    from pyfect.channel import _repeat

    return _repeat.retry(self, policy)

map(f) -> Channel[B, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co]

def map[B](
    self, f: Callable[[OutElem_co], B]
) -> Channel[B, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co]:
    """Apply ``f`` to each item written."""
    from pyfect.channel import _combinators

    return _combinators.map_(self, f)

map_effect(f, *, concurrency=None, unordered=False) -> Channel[B, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def map_effect[B, E2 = Never, R2 = Never](
    self,
    f: Callable[[OutElem_co], Effect[B, E2, R2]],
    *,
    concurrency: Concurrency = None,
    unordered: bool = False,
) -> Channel[
    B, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2
]:
    """Write what ``f(item)`` succeeds with, failing with its error.
    With ``concurrency`` above 1, up to that many run at once; their
    results keep upstream's order unless ``unordered``."""
    from pyfect.channel import _combinators

    return _combinators.map_effect(self, f, concurrency, unordered=unordered)

tap(f, *, concurrency=None) -> Channel[OutElem_co, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def tap[E2 = Never, R2 = Never](
    self, f: Callable[[OutElem_co], Effect[object, E2, R2]], *, concurrency: Concurrency = None
) -> Channel[
    OutElem_co,
    OutErr_co | E2,
    OutDone_co,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """Run ``f`` on each item, then write the item."""
    from pyfect.channel import _combinators

    return _combinators.map_effect(self, lambda a: f(a).as_(a), concurrency)

map_done(f) -> Channel[OutElem_co, OutErr_co, D2, InElem_contra, InErr_contra, InDone_contra, Env_co]

def map_done[D2](
    self, f: Callable[[OutDone_co], D2]
) -> Channel[OutElem_co, OutErr_co, D2, InElem_contra, InErr_contra, InDone_contra, Env_co]:
    """Apply ``f`` to the value the channel ends with."""
    from pyfect.channel import _combinators

    return _combinators.map_done(self, f)

map_done_effect(f) -> Channel[OutElem_co, OutErr_co | E2, D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def map_done_effect[D2, E2 = Never, R2 = Never](
    self, f: Callable[[OutDone_co], Effect[D2, E2, R2]]
) -> Channel[
    OutElem_co, OutErr_co | E2, D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2
]:
    """End with what ``f(done)`` succeeds with, or fail with its error."""
    from pyfect.channel import _combinators

    return _combinators.map_done_effect(self, f)

map_input(f) -> Channel[OutElem_co, OutErr_co, OutDone_co, I2, InErr_contra, InDone_contra, Env_co | R2]

def map_input[I2, R2 = Never](
    self, f: Callable[[I2], Effect[InElem_contra, InErr_contra, R2]]
) -> Channel[OutElem_co, OutErr_co, OutDone_co, I2, InErr_contra, InDone_contra, Env_co | R2]:
    """Read upstream items through ``f``."""
    from pyfect.channel import _combinators

    return _combinators.map_input(self, f)

map_input_error(f) -> Channel[OutElem_co, OutErr_co, OutDone_co, InElem_contra, IE2, InDone_contra, Env_co | R2]

def map_input_error[IE2, R2 = Never](
    self, f: Callable[[IE2], Effect[InErr_contra, InErr_contra, R2]]
) -> Channel[OutElem_co, OutErr_co, OutDone_co, InElem_contra, IE2, InDone_contra, Env_co | R2]:
    """Read upstream's failures through ``f``; the channel sees the
    error ``f`` succeeds or fails with."""
    from pyfect.channel import _combinators

    return _combinators.map_input_error(self, f)

flat_map(f, *, concurrency=None, buffer_size=16) -> Channel[O2, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def flat_map[O2, E2 = Never, R2 = Never](
    self,
    f: Callable[
        [OutElem_co], Channel[O2, E2, Any, InElem_contra, InErr_contra, InDone_contra, R2]
    ],
    *,
    concurrency: Concurrency = None,
    buffer_size: int = 16,
) -> Channel[
    O2, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2
]:
    """Write the items of ``f(item)`` for each item, one channel after
    another; with ``concurrency`` above 1, up to that many at once,
    their items merged through a buffer of ``buffer_size``."""
    from pyfect.channel import _combinators, _merge

    if _combinators.is_sequential(concurrency):
        return _combinators.flat_map(self, f)
    return _merge.merge_all(_combinators.map_(self, f), concurrency, buffer_size, switch=False)

flatten() -> Channel[O2, E | E2, D, I, IE, ID, R | R2]

def flatten[O2, E, D, I, IE, ID, R, E2 = Never, R2 = Never](
    self: TransformMethods[Channel[O2, E2, Any, I, IE, ID, R2], E, D, I, IE, ID, R],
) -> Channel[O2, E | E2, D, I, IE, ID, R | R2]:
    """Write the items of each channel this channel writes, in turn."""
    from pyfect.channel import _combinators

    return _combinators.flat_map(self, _identity)

flatten_array() -> Channel[A, E, D, I, IE, ID, R]

def flatten_array[A, E, D, I, IE, ID, R](
    self: TransformMethods[Sequence[A], E, D, I, IE, ID, R],
) -> Channel[A, E, D, I, IE, ID, R]:
    """Write the items of each sequence one by one."""
    from pyfect.channel import _filters

    return _filters.flatten_array(self)

flatten_take() -> Channel[tuple[A, ...], E | E2, D2, I, IE, ID, R]

def flatten_take[A, E, D, I, IE, ID, R, E2 = Never, D2 = None](
    self: TransformMethods[Take[A, E2, D2], E, D, I, IE, ID, R],
) -> Channel[tuple[A, ...], E | E2, D2, I, IE, ID, R]:
    """Replay each ``Take``: write its items, or fail or end as it says."""
    from pyfect.channel import _combinators
    from pyfect.take import to_pull

    return _combinators.map_effect(self, to_pull, None)

drain() -> Channel[Never, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co]

def drain(
    self,
) -> Channel[Never, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co]:
    """Run the channel for its effects, writing nothing."""
    from pyfect.channel import _combinators

    return _combinators.drain(self)

filter(predicate) -> Channel[Any, Any, Any, Any, Any, Any, Any]

def filter(
    self, predicate: Callable[[Any], bool]
) -> Channel[Any, Any, Any, Any, Any, Any, Any]:
    """Write only the items that pass ``predicate``."""
    from pyfect.channel import _filters

    return _filters.filter_(self, predicate)

filter_effect(predicate) -> Channel[OutElem_co, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def filter_effect[E2 = Never, R2 = Never](
    self, predicate: Callable[[OutElem_co], Effect[bool, E2, R2]]
) -> Channel[
    OutElem_co,
    OutErr_co | E2,
    OutDone_co,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """Write only the items whose ``predicate`` succeeds with ``True``."""
    from pyfect.channel import _filters

    return _filters.filter_effect(self, predicate)

filter_array(predicate) -> Channel[tuple[A, ...], E, D, I, IE, ID, R]

def filter_array[A, E, D, I, IE, ID, R](
    self: TransformMethods[tuple[A, ...], E, D, I, IE, ID, R], predicate: Callable[[A], bool]
) -> Channel[tuple[A, ...], E, D, I, IE, ID, R]:
    """Keep the items of each tuple that pass ``predicate``, skipping
    tuples left empty."""
    from pyfect.channel import _filters

    return _filters.filter_array(self, predicate)

filter_array_effect(predicate) -> Channel[tuple[A, ...], E | E2, D, I, IE, ID, R | R2]

def filter_array_effect[A, E, D, I, IE, ID, R, E2 = Never, R2 = Never](
    self: TransformMethods[tuple[A, ...], E, D, I, IE, ID, R],
    predicate: Callable[[A], Effect[bool, E2, R2]],
) -> Channel[tuple[A, ...], E | E2, D, I, IE, ID, R | R2]:
    """``filter_array`` with an effectful ``predicate``."""
    from pyfect.channel import _filters

    return _filters.filter_array_effect(self, predicate)

filter_map(f) -> Channel[B, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co]

def filter_map[B](
    self, f: Callable[[OutElem_co], Result[B, object]]
) -> Channel[B, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co]:
    """Write ``value`` for each item ``f`` maps to ``Ok(value)``; skip
    the items it maps to ``Err``."""
    from pyfect.channel import _filters

    return _filters.filter_map(self, f)

filter_map_effect(f) -> Channel[B, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def filter_map_effect[B, E2 = Never, R2 = Never](
    self, f: Callable[[OutElem_co], Effect[Result[B, object], E2, R2]]
) -> Channel[
    B, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2
]:
    """``filter_map`` with an effectful ``f``."""
    from pyfect.channel import _filters

    return _filters.filter_map_effect(self, f)

filter_map_array(f) -> Channel[tuple[B, ...], E, D, I, IE, ID, R]

def filter_map_array[A, B, E, D, I, IE, ID, R](
    self: TransformMethods[tuple[A, ...], E, D, I, IE, ID, R],
    f: Callable[[A], Result[B, object]],
) -> Channel[tuple[B, ...], E, D, I, IE, ID, R]:
    """``filter_map`` on the items of each tuple, skipping tuples left
    empty."""
    from pyfect.channel import _filters

    return _filters.filter_map_array(self, f)

filter_map_array_effect(f) -> Channel[tuple[B, ...], E | E2, D, I, IE, ID, R | R2]

def filter_map_array_effect[A, B, E, D, I, IE, ID, R, E2 = Never, R2 = Never](
    self: TransformMethods[tuple[A, ...], E, D, I, IE, ID, R],
    f: Callable[[A], Effect[Result[B, object], E2, R2]],
) -> Channel[tuple[B, ...], E | E2, D, I, IE, ID, R | R2]:
    """``filter_map_array`` with an effectful ``f``."""
    from pyfect.channel import _filters

    return _filters.filter_map_array_effect(self, f)

scan(initial, f) -> Channel[S, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co]

def scan[S](
    self, initial: S, f: Callable[[S, OutElem_co], S]
) -> Channel[S, OutErr_co, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co]:
    """Write ``initial``, then each state ``f`` folds the items into."""
    from pyfect.channel import _filters
    from pyfect.effect import sync

    return _filters.scan_effect(self, initial, lambda s, a: sync(lambda: f(s, a)))

scan_effect(initial, f) -> Channel[S, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def scan_effect[S, E2 = Never, R2 = Never](
    self, initial: S, f: Callable[[S, OutElem_co], Effect[S, E2, R2]]
) -> Channel[
    S, OutErr_co | E2, OutDone_co, InElem_contra, InErr_contra, InDone_contra, Env_co | R2
]:
    """``scan`` with an effectful ``f``."""
    from pyfect.channel import _filters

    return _filters.scan_effect(self, initial, f)

map_accum(initial, f, *, on_halt=None) -> Channel[Any, Any, Any, Any, Any, Any, Any]

def map_accum(
    self,
    initial: Callable[[], Any],
    f: Callable[[Any, Any], Any],
    *,
    on_halt: Callable[[Any], Sequence[Any]] | None = None,
) -> Channel[Any, Any, Any, Any, Any, Any, Any]:
    """Thread a state through the items: ``f(state, item)`` gives the
    next state and the items to write (or an effect of both). When
    upstream stops, ``on_halt(state)`` gives the last items to write."""
    from pyfect.channel import _filters

    return _filters.map_accum(self, initial, f, on_halt)

concat_with(f) -> Channel[OutElem_co | O2, OutErr_co | E2, D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def concat_with[O2, E2 = Never, D2 = None, R2 = Never](
    self,
    f: Callable[
        [OutDone_co], Channel[O2, E2, D2, InElem_contra, InErr_contra, InDone_contra, R2]
    ],
) -> Channel[
    OutElem_co | O2, OutErr_co | E2, D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2
]:
    """When this channel ends, go on with ``f(done)`` on the same
    upstream. This channel's resources are released first."""
    from pyfect.channel import _combinators

    return _combinators.concat_with(self, f)

concat(that) -> Channel[OutElem_co | O2, OutErr_co | E2, D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def concat[O2, E2 = Never, D2 = None, R2 = Never](
    self, that: Channel[O2, E2, D2, InElem_contra, InErr_contra, InDone_contra, R2]
) -> Channel[
    OutElem_co | O2, OutErr_co | E2, D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2
]:
    """Write this channel's items, then ``that``'s, ending as ``that``."""
    from pyfect.channel import _combinators

    require_channel(that, "that")
    return _combinators.concat_with(self, lambda _: that)

or_else_if_empty(f) -> Channel[OutElem_co | O2, OutErr_co | E2, OutDone_co | D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def or_else_if_empty[O2, E2 = Never, D2 = None, R2 = Never](
    self,
    f: Callable[
        [OutDone_co], Channel[O2, E2, D2, InElem_contra, InErr_contra, InDone_contra, R2]
    ],
) -> Channel[
    OutElem_co | O2,
    OutErr_co | E2,
    OutDone_co | D2,
    InElem_contra,
    InErr_contra,
    InDone_contra,
    Env_co | R2,
]:
    """Go on with ``f(done)`` if this channel ends without writing."""
    from pyfect.channel import _combinators

    return _combinators.or_else_if_empty(self, f)

combine(that, initial, f) -> Channel[Any, Any, Any, Any, Any, Any, Any]

def combine(
    self,
    that: Channel[Any, Any, Any, Any, Any, Any, Any],
    initial: Callable[[], Any],
    f: Callable[[Any, Any, Any], Effect[tuple[Any, Any], Any, Any]],
) -> Channel[Any, Any, Any, Any, Any, Any, Any]:
    """Write what ``f(state, left, right)`` makes from the two pulls,
    threading its state. ``f`` ends the channel by failing with
    ``Done``."""
    from pyfect.channel import _combinators

    require_channel(that, "that")
    return _combinators.combine(self, that, initial, f)

pipe_to(that) -> Channel[O2, E2, D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def pipe_to[O2, E2 = Never, D2 = None, R2 = Never](
    self, that: Channel[O2, E2, D2, OutElem_co, OutErr_co, OutDone_co, R2]
) -> Channel[O2, E2, D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]:
    """Feed this channel's output to ``that`` as its upstream."""
    from pyfect.channel import _combinators

    require_channel(that, "that")
    return _combinators.pipe_to(self, that)

pipe_to_or_fail(that) -> Channel[O2, OutErr_co | E2, D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]

def pipe_to_or_fail[O2, E2 = Never, D2 = None, R2 = Never](
    self, that: Channel[O2, E2, D2, OutElem_co, Never, OutDone_co, R2]
) -> Channel[O2, OutErr_co | E2, D2, InElem_contra, InErr_contra, InDone_contra, Env_co | R2]:
    """``pipe_to`` for a ``that`` that cannot see upstream failures: they
    pass through ``that`` untouched (its error handlers never see them)
    and fail the channel."""
    from pyfect.channel import _upstream

    require_channel(that, "that")
    return _upstream.pipe_to_or_fail(self, that)