stream

from pyfect import stream

fx.run_sync(stream.range(0, 5).run_collect())  # [0, 1, 2, 3, 4]

catch_if_refinement(refinement, f, *, or_else=None) -> Any

def catch_if_refinement(
    refinement: Callable[[Any], bool], f: Handler, *, or_else: Handler | None = None
) -> Any:
    """Recover from the typed errors ``refinement`` accepts and remove them
    from the stream's errors. For a plain ``bool`` predicate, use the
    ``catch_if`` method."""
    if or_else is None:
        return _CatchSome[Any, Any, Any, Any](refinement, f)
    return _CatchAll[Any, Any, Any, Any, Any, Any](refinement, f, or_else)

catch_tag(tag, f, *, or_else=None) -> Any

def catch_tag(
    tag: type[Any] | tuple[type[Any], ...], f: Handler, *, or_else: Handler | None = None
) -> Any:
    """Recover from the errors that are instances of ``tag`` (a class or a
    tuple of up to three) and remove them from the stream's errors."""
    predicate: Callable[[Any], bool] = lambda e: isinstance(e, tag)  # noqa: E731
    if or_else is None:
        return _CatchSome[Any, Any, Any, Any](predicate, f)
    return _CatchAll[Any, Any, Any, Any, Any, Any](predicate, f, or_else)

catch_reason(tag, reason, f, *, or_else=None) -> Any

def catch_reason(
    tag: type[Any],
    reason: type[Any],
    f: ReasonHandler[AnyStream],
    *,
    or_else: ReasonHandler[AnyStream] | None = None,
) -> Any:
    """Switch to ``f(reason, error)`` when the stream fails with a ``tag``
    error whose ``reason`` is an instance of ``reason``; with ``or_else``,
    other ``tag`` errors switch to ``or_else(reason, error)``."""
    run = _reason_catch(tag, ((reason, f),), or_else)
    return (
        _CatchReason[Any, Any, Any](run)
        if or_else is None
        else _CatchReasonAll[Any, Any, Any, Any, Any, Any, Any, Any](run)
    )

catch_reasons(tag, *cases, *, or_else=None) -> Any

def catch_reasons(
    tag: type[Any],
    *cases: tuple[type[Any], ReasonHandler[AnyStream]],
    or_else: ReasonHandler[AnyStream] | None = None,
) -> Any:
    """``catch_reason`` with a ``(reason_class, handler)`` pair per reason,
    tried in order."""
    run = _reason_catch(tag, cases, or_else)
    return (
        _CatchReason[Any, Any, Any](run)
        if or_else is None
        else _CatchReasonAll[Any, Any, Any, Any, Any, Any, Any, Any](run)
    )

unwrap_reason(tag) -> _UnwrapReason[T, T]

def unwrap_reason[T: HasReason[Any]](tag: type[T]) -> _UnwrapReason[T, T]:
    """Fail with the ``reason`` of a ``tag`` error instead of the error."""
    return _UnwrapReason(_reason_catch(tag, (), lambda reason, _: fail(reason)))

die(defect) -> Stream[Never]

def die(defect: BaseException) -> Stream[Never]:
    """A stream failing with the defect ``defect``."""
    return Stream(channel.die(defect))

fail(error) -> Stream[Never, E]

def fail[E](error: E) -> Stream[Never, E]:
    """A stream failing with ``error``."""
    return Stream(channel.fail(error))

fail_cause(cause) -> Stream[Never, E]

def fail_cause[E](cause: Cause[E]) -> Stream[Never, E]:
    """A stream failing with ``cause``."""
    return Stream(channel.fail_cause(cause))

fail_cause_sync(thunk) -> Stream[Never, E]

def fail_cause_sync[E](thunk: Callable[[], Cause[E]]) -> Stream[Never, E]:
    """A stream failing with ``thunk()``, called each time it runs."""
    return Stream(channel.fail_cause_sync(thunk))

fail_sync(thunk) -> Stream[Never, E]

def fail_sync[E](thunk: Callable[[], E]) -> Stream[Never, E]:
    """A stream failing with ``thunk()``, called each time it runs."""
    return Stream(channel.fail_sync(thunk))

from_arrays(*arrays) -> Stream[A]

def from_arrays[A](*arrays: Iterable[A]) -> Stream[A]:
    """A stream of the items of ``arrays``, one chunk per non-empty array."""
    chunks = tuple(chunk for chunk in builtins.map(tuple, arrays) if chunk)
    return Stream(channel.from_iterable(chunks))

from_effect(eff) -> Stream[A, E, R]

def from_effect[A, E = Never, R = Never](eff: Effect[A, E, R]) -> Stream[A, E, R]:
    """A stream of the one item ``eff`` succeeds with."""
    require_effect(eff, "eff")
    return Stream(channel.from_effect(eff.map(_one)))

from_effect_drain(eff) -> Stream[Never, E, R]

def from_effect_drain[A, E = Never, R = Never](eff: Effect[A, E, R]) -> Stream[Never, E, R]:
    """A stream that runs ``eff`` and ends, writing nothing."""
    require_effect(eff, "eff")
    return Stream(channel.from_effect_drain(eff))

from_effect_repeat(eff) -> Stream[Any, Any, Any]

def from_effect_repeat(eff: Effect[Any, Any, Any]) -> Stream[Any, Any, Any]:
    """A stream of what ``eff`` succeeds with, run again for each item. It
    ends when ``eff`` fails with ``Done()``."""
    require_effect(eff, "eff")
    return from_pull(_Succeed(eff.map(_one)))

from_effect_schedule(eff, schedule) -> Stream[A, EE, RR]

def from_effect_schedule[A, EE = Never, RR = Never](
    eff: Effect[A, EE, RR], schedule: Schedule[object, A, EE, RR]
) -> Stream[A, EE, RR]:
    """A stream of what ``eff`` succeeds with, run again after each delay
    of ``schedule`` (stepped with the last value) until it stops. Each run
    of ``eff`` sees the step's ``schedule.CurrentMetadata``."""
    require_effect(eff, "eff")

    def make_pull(step: Callable[[Any], Effect[Metadata[Any, Any] | Stop[Any], Any, Any]]) -> Any:
        first = _with_metadata(eff, CurrentMetadata.default())

        def started(value: Any) -> AnyPull:
            last = value

            def again(decision: Metadata[Any, Any] | Stop[Any]) -> AnyPull:
                if isinstance(decision, Stop):
                    return _ENDED
                return _with_metadata(eff, decision).map(keep)

            def keep(next_value: Any) -> tuple[Any]:
                nonlocal last
                last = next_value
                return (next_value,)

            emitted = False

            def pull() -> AnyPull:
                nonlocal emitted
                if not emitted:
                    emitted = True
                    return _Succeed((last,))
                return step(last).flat_map(again)

            return _Succeed(_Suspend(pull))

        return first.flat_map(started)

    return from_pull(to_step_with_metadata(schedule).flat_map(make_pull))

from_iterable(iterable, *, chunk_size=None) -> Stream[A]

def from_iterable[A](iterable: Iterable[A], *, chunk_size: int | None = None) -> Stream[A]:
    """A stream of the items of ``iterable``. A list or a tuple is one
    chunk unless ``chunk_size`` is given; anything else is iterated anew
    each run (a generator or an iterator only once) in chunks of up to
    ``chunk_size`` (``DEFAULT_CHUNK_SIZE`` by default, at least 1)."""
    if isinstance(iterable, list | tuple) and chunk_size is None:
        chunk = tuple(iterable)
        return Stream(channel.succeed(chunk)) if chunk else empty
    size = DEFAULT_CHUNK_SIZE if chunk_size is None else chunk_size
    chunks = channel.from_iterator_array(lambda: iter(iterable), size)
    return Stream(chunks.map_done(lambda _: None))

from_iterable_effect(eff) -> Stream[A, E, R]

def from_iterable_effect[A, E = Never, R = Never](
    eff: Effect[Iterable[A], E, R],
) -> Stream[A, E, R]:
    """A stream of the items of the iterable ``eff`` succeeds with."""
    require_effect(eff, "eff")
    channels: Any = eff.map(_iterable_channel)
    return Stream(channel.unwrap(channels))

from_iterable_effect_repeat(eff) -> Stream[Any, Any, Any]

def from_iterable_effect_repeat(eff: Effect[Any, Any, Any]) -> Stream[Any, Any, Any]:
    """A stream of the items of the iterables ``eff`` succeeds with, run
    again after each. It ends when ``eff`` fails with ``Done()``."""
    require_effect(eff, "eff")
    iterables: Any = channel.from_pull(_Succeed(eff))
    return Stream(iterables.flat_map(_iterable_channel))

iterate(initial, f) -> Stream[A]

def iterate[A](initial: A, f: Callable[[A], A]) -> Stream[A]:
    """The endless stream ``initial``, ``f(initial)``, ``f(f(initial))``, ..."""
    return unfold(initial, lambda a: _Sync(lambda: (a, f(a))))

make(*values) -> Stream[Any]

def make(*values: Any) -> Stream[Any]:
    """A stream of ``values``, in one chunk."""
    return from_iterable(values)

paginate(initial, f) -> Stream[A, E, R]

def paginate[S, A, E = Never, R = Never](
    initial: S, f: Callable[[S], Effect[tuple[Iterable[A], Option[S]], E, R]]
) -> Stream[A, E, R]:
    """A stream of pages: ``f(state)`` succeeds with ``(items, next)``,
    where ``next`` is ``Some(state)`` for another page or ``None`` after
    this one. Empty pages write nothing."""

    def make_pull() -> AnyPull:
        state = initial
        done = False

        def loop() -> AnyPull:
            return _ENDED if done else f(state).flat_map(written)

        def written(page: tuple[Iterable[A], Option[S]]) -> AnyPull:
            nonlocal state, done
            items, following = page
            if following is None:
                done = True
            else:
                state = following.value
            chunk = tuple(items)
            return _Succeed(chunk) if chunk else _Suspend(loop)

        return _Suspend(loop)

    return from_pull(_Sync(make_pull))

range(start, stop, *, chunk_size=DEFAULT_CHUNK_SIZE) -> Stream[int]

def range(start: int, stop: int, *, chunk_size: int = DEFAULT_CHUNK_SIZE) -> Stream[int]:
    """The integers from ``start`` up to, not including, ``stop``, as
    Python's ``range`` (Effect's ``range`` includes its end: see
    ``range_inclusive``). Chunks hold up to ``chunk_size`` (at least 1)."""
    if start >= stop:
        return empty
    size = max(1, chunk_size)

    def make_pull() -> AnyPull:
        current = start

        def pull() -> AnyPull:
            nonlocal current
            if current >= stop:
                return _ENDED
            end = min(current + size, stop)
            chunk = tuple(builtins.range(current, end))
            current = end
            return _Succeed(chunk)

        return _Suspend(pull)

    return from_pull(_Sync(make_pull))

range_inclusive(minimum, maximum, *, chunk_size=DEFAULT_CHUNK_SIZE) -> Stream[int]

def range_inclusive(
    minimum: int, maximum: int, *, chunk_size: int = DEFAULT_CHUNK_SIZE
) -> Stream[int]:
    """The integers from ``minimum`` to ``maximum``, both included
    (Effect's ``range``); empty when ``minimum > maximum``."""
    return range(minimum, maximum + 1, chunk_size=chunk_size)

scoped(stream) -> Stream[Any, Any, Any]

def scoped(stream: AnyStream) -> Stream[Any, Any, Any]:
    """Give the stream a ``Scope`` of its own, closed as soon as it fails
    or ends (or its run ends), and remove ``Scope`` from its services."""
    require_stream(stream, "stream")
    scoped_channel: Any = channel.scoped(stream._channel)
    return Stream(scoped_channel)

service(tag) -> Stream[S, Never, T]

def service[S, T](tag: TagLike[S, T]) -> Stream[S, Never, T]:
    """A stream of the one item: the service ``tag`` names."""
    return from_effect(tag())

service_option(tag) -> Stream[Option[S]]

def service_option[S, T](tag: TagLike[S, T]) -> Stream[Option[S]]:
    """A stream of the one item ``Some(service)`` if ``tag``'s service is
    provided, else ``None``."""
    return from_effect(_service_option(tag))

succeed(value) -> Stream[A]

def succeed[A](value: A) -> Stream[A]:
    """A stream of the one item ``value``."""
    return Stream(channel.succeed((value,)))

suspend(thunk) -> Stream[A, E, R]

def suspend[A, E, R](thunk: Callable[[], StreamBase[A, E, R]]) -> Stream[A, E, R]:
    """The stream ``thunk()`` returns, built each time it runs."""
    return SuspendedStream(thunk)

sync(thunk) -> Stream[A]

def sync[A](thunk: Callable[[], A]) -> Stream[A]:
    """A stream of the one item ``thunk()``, called each time it runs."""
    return Stream(channel.sync(lambda: (thunk(),)))

unfold(initial, f) -> Stream[A, E, R]

def unfold[S, A, E = Never, R = Never](
    initial: S, f: Callable[[S], Effect[tuple[A, S] | None, E, R]]
) -> Stream[A, E, R]:
    """A stream of the items ``f`` makes from a state: ``f(state)``
    succeeds with ``(item, next_state)``, or ``None`` to end."""

    def make_pull() -> AnyPull:
        state = initial

        def written(step: tuple[A, S] | None) -> AnyPull:
            nonlocal state
            if step is None:
                return _ENDED
            item, state = step
            return _Succeed((item,))

        return _Suspend(lambda: f(state)).flat_map(written)

    return from_pull(_Sync(make_pull))

unwrap(eff) -> Stream[Any, Any, Any]

def unwrap(eff: Effect[AnyStream, Any, Any]) -> Stream[Any, Any, Any]:
    """The stream ``eff`` succeeds with, built on the first pull. ``eff``
    runs with the stream's ``Scope``, so its resources live as long as the
    stream's."""
    require_effect(eff, "eff")
    channels: Any = eff.map(lambda stream: stream._channel)
    return from_channel(channel.unwrap(channels))

provide(source, /, *, local=False) -> _ProvideLayer[Any, Any, Any] | _ProvideContext[Any]

def provide(
    source: Layer[Any, Any, Any] | Context[Any], /, *, local: bool = False
) -> _ProvideLayer[Any, Any, Any] | _ProvideContext[Any]:
    """Provide a context's services, or build a layer each time the stream
    runs (its resources released when the run ends) and provide what it
    builds. Layers an enclosing ``provide`` built are reused unless
    ``local``."""
    if isinstance(source, Context):
        return _ProvideContext(channel.provide_context(source))
    require_layer(source, "source", "a Layer or a Context")
    return _ProvideLayer(channel.provide(source, local=local))

provide_context(ctx) -> _ProvideContext[RO]

def provide_context[RO](ctx: Context[RO]) -> _ProvideContext[RO]:
    """Provide every service of ``ctx`` to a stream and remove them from
    its services."""
    return _ProvideContext(channel.provide_context(ctx))

provide_service(tag, service) -> _ProvideService[T]

def provide_service[S, T](tag: TagLike[S, T], service: S) -> _ProvideService[T]:
    """Provide one service to a stream and remove it from its services::

    users.pipe(stream.provide_service(Database, InMemoryDb()))
    """
    return _ProvideService(channel.provide_service(tag, service))

provide_service_effect(tag, acquire) -> _ProvideServiceEffect[T, E2, R2]

def provide_service_effect[S, T, E2 = Never, R2 = Never](
    tag: TagLike[S, T], acquire: Effect[S, E2, R2]
) -> _ProvideServiceEffect[T, E2, R2]:
    """Run ``acquire`` each time the stream runs and provide its value as
    ``tag``'s service, removing ``tag`` from the stream's services."""
    require_effect(acquire, "acquire")
    return _ProvideServiceEffect(channel.provide_service_effect(tag, acquire))

update_context(f) -> _UpdateContext[R2, RO]

def update_context[R2, RO](f: Callable[[Context[R2]], Context[RO]]) -> _UpdateContext[R2, RO]:
    """Build the stream with the services ``f(context)`` instead of
    ``context``, so it needs ``R2`` instead of what it needed."""
    update: Callable[[Context[Any]], Context[Any]] = f
    return _UpdateContext(lambda ch: update_context_with(ch, update))

with_execution_plan(plan, *, prevent_fallback_on_partial_stream=False, on_event=None) -> _WithExecutionPlan[Any, Any, Any, Any]

def with_execution_plan(
    plan: ExecutionPlan[Any, Any, Any, Any],
    *,
    prevent_fallback_on_partial_stream: bool = False,
    on_event: OnEvent[Any, Any] | None = None,
) -> _WithExecutionPlan[Any, Any, Any, Any]:
    """Run a stream under ``plan``, as ``pyfect.with_execution_plan`` runs
    an effect: a failing step falls back to the next, even after writing
    items, unless ``prevent_fallback_on_partial_stream`` (then it fails).
    Use with ``.pipe``. A downstream consumer that stops early ends the
    attempt with ``AttemptSuccess``."""
    return _WithExecutionPlan(plan, prevent_fallback_on_partial_stream, on_event)

merge_all(streams, *, concurrency, buffer_size=16) -> Stream[A, EE, RR]

def merge_all[A, EE = Never, RR = Never](
    streams: Iterable[StreamBase[A, EE, RR]], *, concurrency: Concurrency, buffer_size: int = 16
) -> Stream[A, EE, RR]:
    """Run ``streams``, up to ``concurrency`` at once, writing their items
    as they come (at most ``buffer_size`` waiting)."""
    from pyfect.stream._constructors import from_iterable

    parts = list(streams)
    for part in parts:
        require_stream(part, "streams")
    return from_iterable(parts).flatten(concurrency=concurrency, buffer_size=buffer_size)

race_all(*streams) -> Stream[A, EE, RR]

def race_all[A, EE = Never, RR = Never](*streams: StreamBase[A, EE, RR]) -> Stream[A, EE, RR]:
    """The stream that writes first; the others are interrupted."""
    from pyfect.stream import _zipping

    require_streams(streams, "streams")
    return _zipping.race_all(*streams)

zip_latest_all(*streams) -> Stream[Any, Any, Any]

def zip_latest_all(*streams: StreamBase[Any, Any, Any]) -> Stream[Any, Any, Any]:
    """Once every stream has written, write a tuple of each one's latest
    item whenever any writes; end when any ends (at once if one ends
    without writing)."""
    from pyfect.stream import _constructors, _zipping

    require_streams(streams, "streams")
    if not streams:
        return _constructors.empty
    return _zipping.zip_latest_all(*streams)

callback(f, *, buffer_size=None, strategy='suspend') -> Stream[Any, Any, Any]

def callback(
    f: Callable[[Queue[Any, Any]], Effect[object, Any, Any]],
    *,
    buffer_size: int | None = None,
    strategy: Strategy = "suspend",
) -> Stream[Any, Any, Any]:
    """A stream of what ``f`` offers to a queue. ``f`` runs on a fiber, with
    the stream's ``Scope``, when the stream starts; the stream ends when the
    queue ends and fails with ``f``'s failure or the queue's. With a
    ``buffer_size`` the queue is bounded and ``strategy`` says what a full
    one does (``"suspend"`` waits: backpressure)."""
    chunks: Any = channel.callback_array(f, buffer_size=buffer_size, strategy=strategy)
    return Stream(chunks)

from_async_iterable(iterable, on_error) -> Stream[A, E]

def from_async_iterable[A, E](
    iterable: AsyncIterable[A], on_error: Callable[[Exception], E]
) -> Stream[A, E]:
    """A stream of what ``iterable`` yields, failing with
    ``on_error(exception)`` if it raises. The iterator is closed when the
    run ends. Needs an asyncio loop (``run_async``)."""
    return Stream(channel.from_async_iterable_array(iterable, on_error))

from_pubsub(pubsub) -> Stream[A]

def from_pubsub[A](pubsub: PubSub[A]) -> Stream[A]:
    """A stream of what is published to ``pubsub`` after the stream starts
    (and what it replays); it ends when the pubsub shuts down."""
    return Stream(channel.from_pubsub_array(pubsub))

from_pubsub_take(pubsub) -> Stream[A, E]

def from_pubsub_take[A, E](pubsub: PubSub[Take[A, E, None]]) -> Stream[A, E]:
    """A stream replaying the ``Take`` values published to ``pubsub``: the
    chunks, then the failure or the end."""
    return Stream(channel.from_pubsub_take(pubsub))

from_queue(queue) -> Stream[Any, Any]

def from_queue(queue: Dequeue[Any, Any]) -> Stream[Any, Any]:
    """A stream taking everything waiting in ``queue`` at each pull, until
    it ends (or fails)."""
    return Stream(channel.from_queue_array(queue))

from_schedule(schedule) -> Stream[O, E, R]

def from_schedule[O, E = Never, R = Never](schedule: Schedule[O, object, E, R]) -> Stream[O, E, R]:
    """A stream of the outputs of ``schedule``, each after its delay; it
    ends when the schedule stops."""

    def with_step(step: Callable[[None], Effect[Metadata[O, Any] | Stop[O], E, R]]) -> AnyPull:
        def written(decision: Metadata[O, Any] | Stop[O]) -> AnyPull:
            return end_with(None) if isinstance(decision, Stop) else _Succeed((decision.output,))

        return step(None).flat_map(written)

    return from_pull(to_step_with_metadata(schedule).map(with_step))

from_subscription(subscription) -> Stream[A]

def from_subscription[A](subscription: Subscription[A]) -> Stream[A]:
    """A stream of the messages of ``subscription``; it ends when the
    pubsub shuts down."""
    return Stream(channel.from_subscription_array(subscription))

tick(interval) -> Stream[None]

def tick(interval: Duration) -> Stream[None]:
    """An endless stream of ``None``: the first at once, then one after
    each ``interval``."""
    seconds = to_seconds(interval)
    tock: Effect[tuple[None]] = _Succeed((None,))

    def make_pull() -> AnyPull:
        delayed = tock.delay(seconds)
        first = True

        def pull() -> AnyPull:
            nonlocal first
            if first:
                first = False
                return tock
            return delayed

        return _Suspend(pull)

    return from_pull(_Sync(make_pull))

from_channel(channel) -> Stream[A, E, R]

def from_channel[A, E = Never, R = Never](
    channel: Channel[tuple[A, ...], E, None, object, object, object, R],
) -> Stream[A, E, R]:
    """The stream of the items in the tuples ``channel`` writes. Each tuple
    must be non-empty."""
    require_channel(channel, "channel")
    return Stream(channel)

from_pull(make) -> Stream[Any, Any, Any]

def from_pull(make: Effect[Any, Any, Any]) -> Stream[Any, Any, Any]:
    """A stream whose pull ``make`` builds each time it runs. The pull
    succeeds with a non-empty tuple, fails, or ends with ``Done()``."""
    require_effect(make, "make")
    return Stream(channel_from_pull(make))

to_channel(stream) -> StreamChannel[A, E, R]

def to_channel[A, E, R](stream: StreamBase[A, E, R]) -> StreamChannel[A, E, R]:
    """The channel a stream is made of: it writes the stream's chunks."""
    require_stream(stream, "stream")
    return stream._channel

transform_pull(stream, f) -> Stream[Any, Any, Any]

def transform_pull(
    stream: StreamBase[Any, Any, Any],
    f: Callable[[Any, CloseableScope], Effect[Any, Any, Any]],
) -> Stream[Any, Any, Any]:
    """A stream whose pull is ``f(pull, scope)`` of ``stream``'s pull. The
    upstream pull is built in the new stream's scope."""
    require_stream(stream, "stream")
    channel = stream._channel

    def transform(_upstream: Any, scope: CloseableScope) -> Effect[Any, Any, Any]:
        return channel.to_pull_scoped(scope).flat_map(lambda pull: f(pull, scope))

    built: Channel[Any, Any, Any, Any, Any, Any, Any] = from_transform(transform)
    return Stream(built)

transform_pull_bracket(stream, f) -> Stream[Any, Any, Any]

def transform_pull_bracket(
    stream: StreamBase[Any, Any, Any],
    f: Callable[[Any, CloseableScope, CloseableScope], Effect[Any, Any, Any]],
) -> Stream[Any, Any, Any]:
    """``transform_pull`` with a third argument: a child of the stream's
    scope, closed as soon as the new stream fails or ends."""
    require_stream(stream, "stream")
    channel = stream._channel

    def transform(
        _upstream: Any, scope: CloseableScope, forked: CloseableScope
    ) -> Effect[Any, Any, Any]:
        return channel.to_pull_scoped(scope).flat_map(lambda pull: f(pull, scope, forked))

    built: Channel[Any, Any, Any, Any, Any, Any, Any] = from_transform_bracket(transform)
    return Stream(built)