aioreactive

aioreactive is an asynchronous and reactive Python library for asyncio using async and await. It is built on the Expression functional library and provides a unification of RxPY and reactive programming with the asyncio event loop.

Key Design Goals:

• Python 3.10+ only: Requires Python 3.10 or higher due to a dependency on Expression v5.
• Async-first: All operators, sending values, subscribing, and disposing are async operations.
• Single Scheduler: Everything runs on the asyncio base event-loop; no multi-threading is used by default.
• Back-pressure: Uses implicit synchronous back-pressure where producers await downstream consumers.
• Functional & OO: Supports both functional pipe-forward programming and object-oriented method chaining.

An async stream is an object that acts as both an AsyncObserver and an AsyncObservable. You can create them explicitly using:

• AsyncSubject (Alias for AsyncMultiStream): Supports multiple observers. It is 'hot', meaning it will drop any event sent if there are currently no observers attached.
• AsyncSingleStream: Supports a single observer. It is 'cold', meaning it will await any producer until an observer is attached.

Example of using an AsyncSubject as a stream:

import aioreactive as rx

stream = rx.AsyncSubject()  # Alias for AsyncMultiStream

sink = rx.AsyncAnonymousObserver()
await stream.subscribe_async(sink)
await stream.asend(42)

aioreactive is built around two core asynchronous abstractions that are the duals of AsyncIterable and AsyncIterator:

1. AsyncObservable: A producer of events. It provides the subscribe_async(observer) method to attach an observer.
2. AsyncObserver: A consumer of events. It implements three primary async methods:
   • asend(value): Receives a new value.
   • athrow(error): Receives an error.
   • aclose(): Signals the end of the stream.

To avoid implementing a full class every time, you can use AsyncAnonymousObserver, which constructs an observer from plain async functions.

# Basic pattern for subscribing an observer to an observable
subscription = await observable.subscribe_async(observer)

While related, AsyncIterable and AsyncObservable serve different paradigms:

• AsyncIterable (Pull-based): An async iterable world where the consumer requests data. It is often simpler for basic operations like map() and filter().
• AsyncObservable (Push-based): A reactive world where data is pushed to the consumer. This is superior when dealing with time-based operations, such as the delay() operator.

aioreactive allows you to convert between these two worlds, making it easy to use reactive operators and then switch to an async iterable just before consumption.

Instead of using the pipe function, you can use a fluent method-chaining style. When an AsyncObservable is created via methods like AsyncRx.from_iterable(), it returns an AsyncChainedObservable. This allows you to chain operators like .filter() and .map() directly onto the observable object.

from aioreactive import AsyncRx, AsyncAnonymousObserver

# Create an observable and chain operators
xs = AsyncRx.from_iterable([1, 2, 3])

async def mapper(value):
    return value * 10

async def predicate(value):
    return value > 1

# Chaining style
ys = xs.filter(predicate).map(mapper)

# Subscribing to the chained observable
async def on_next(value):
    print(value)

subscription = await ys.subscribe_async(AsyncAnonymousObserver(on_next))
await subscription

You can convert an AsyncObservable into an AsyncIterable using to_async_iterable (or by using AsyncIteratorObserver) to use the async for syntax. This transforms the model from a 'push' model to a 'pull' model.

Because the producer awaits the iterator to pick up the item, this provides built-in back-pressure.

It is recommended to wrap the subscription in an async with block to ensure the subscription lifetime is managed correctly.

import aioreactive as rx

xs = rx.from_iterable([1, 2, 3])
result = []

obv = rx.AsyncIteratorObserver(xs)
# Using async with to control subscription lifetime
async with await xs.subscribe_async(obv) as subscription:
    async for x in obv:
        result.append(x)

assert result == [1, 2, 3]

To start streaming items from an AsyncObservable, you must call subscribe_async(observer). This method returns a disposable subscription object.

To stop receiving events and clean up resources, you must await the dispose_async() method on the subscription.

Use AsyncAnonymousObserver to quickly create an observer from an async function.

# Using an anonymous observer with an async function
async def asend(value):
    print(value)

# Subscribe to the source
disposable = await subscribe_async(source, AsyncAnonymousObserver(asend))

# Later, to unsubscribe:
await disposable.dispose_async()

aioreactive supports a functional 'pipe' programming style via the Expression library. This allows you to compose operators by partially applying them with arguments and passing the source stream as the final argument in a pipe() call.

This is particularly useful for long, readable transformation chains.

import aioreactve as rx
from expression import pipe

async def main():
    stream = rx.AsyncSubject()
    obv = rx.AsyncIteratorObserver()

    # Composing a pipeline using pipe()
    ys = pipe(
        stream,
        rx.map(lambda x: x["term"]),
        rx.filter(lambda text: len(text) > 2),
        rx.debounce(0.75),
        rx.distinct_until_changed(),
        rx.map(search_wikipedia),
        rx.switch_latest(),
    )

    async with await stream.subscribe_async(obv) as subscription:
        async for value in ys:
            print(value)

The AsyncRx class is a wrapper around AsyncObservable that enables a fluent, method-chaining API similar to classic ReactiveX. It allows you to apply operators (like map, filter, debounce) directly as methods on the observable instance.

All methods in AsyncRx are lazily imported to optimize startup performance.

import aioreactive as rx

# Method chaining style
stream = rx.AsyncRx.from_iterable([1, 2, 3]).map(lambda x: x + 2).filter(lambda x: x < 3)

While AsyncRx provides a chained API, you can also use operators as standalone functions with the pipe utility. This is useful when working with raw AsyncObservable objects or when you prefer a functional programming style.

Common pipeable operators include filter, map, flat_map, debounce, and subscribe_async.

import aioreactive as rx
from aioreactive import pipe, filter, map

async def main():
    source = rx.AsyncRx.from_iterable([1, 2, 3, 4, 5])
    
    # Using the pipe function
    pipeline = pipe(
        source,
        filter(lambda x: x % 2 == 0),
        map(lambda x: x * 10)
    )
    
    # To execute, you still need to subscribe
    await pipeline.subscribe_async(send=print)

The Flatten protocol defines a transformation for nested observables. Specifically, it describes a mechanism to project from an observable of observables into a single observable.

A zipping projection follows this shape: AsyncObservable[AsyncObservable[TSource]] -> AsyncObservable[Tuple[TSource, TResult]]

Note that the Flatten protocol itself is defined as a Callable that takes an AsyncObservable[AsyncObservable[_TSource]] and returns an AsyncObservable[_TSource].