Reactive Extensions for Python (RxPY)

3.2.0 · active · verified Thu Apr 09

Reactive Extensions for Python (RxPY) is a library for composing asynchronous and event-based programs using observable sequences and pipable query operators. It enables developers to represent asynchronous data streams with Observables, query them using a rich set of operators, and manage concurrency with Schedulers. The library (via the 'rx' PyPI package) is currently at version 3.2.0, providing a robust framework for reactive programming in Python.

Warnings

breaking The official RxPY project renamed its main module from `rx` to `reactivex` starting with version 4.x. While `pip install rx` currently provides version 3.x, upgrading to the latest RxPY (installed as `reactivex`) will require changes to import statements.
Fix: When migrating to RxPY v4+ (installed via `pip install reactivex`), update your imports from `import rx` to `import reactivex as rx` and `from rx import operators as ops` to `from reactivex import operators as ops`.
breaking In RxPY v4, the `mapper` function argument was removed from operators that combine values from several observables (e.g., `combine_latest`, `group_join`, `join`, `with_latest_from`, `zip`, `zip_with_iterable`).
Fix: Refactor code to apply transformations using separate `map` operators after the combining operation, instead of relying on the `mapper` argument within the combining operator itself.
breaking For developers creating custom operators using functional composition, the global `rx.pipe` function (used as a helper to compose other operators) was renamed to `reactivex.compose` in version 4.x.
Fix: If you defined custom operators using `rx.pipe` (e.g., `def custom_op(): return rx.pipe(ops.map(...))`), change it to `reactivex.compose` when using RxPY v4+ (`import reactivex`).
gotcha When using operators with multiple optional arguments, it is highly recommended to use named keyword arguments rather than positional arguments to prevent unexpected behavior due to argument order or future API changes.
Fix: Always pass optional arguments to operators using their keyword names (e.g., `observable.pipe(ops.buffer_with_time(buffer_time_span=1.0, buffer_time_reset=0.5))`).

Install

pip install rx Install `rx` package (version 3.x)

Imports

rx
```
import rx
```
The PyPI package `rx` (version 3.x) should be imported as `rx`. The `reactivex` import is for the newer v4+ generation of RxPY, typically installed via `pip install reactivex`.
operators
```
from rx import operators as ops
```
Operators are typically imported from the `rx.operators` submodule and aliased as `ops`.

Quickstart

This quickstart demonstrates creating an observable from a simple sequence, applying transformation and filtering operators using the `pipe` method, and subscribing with `on_next`, `on_error`, and `on_completed` handlers. This uses the pipe-based operator chaining introduced in v3.

import rx
from rx import operators as ops

# Create an observable from a sequence of items
source = rx.of("Alpha", "Beta", "Gamma", "Delta", "Epsilon")

# Define a pipeline of operators
composed = source.pipe(
    ops.map(lambda s: len(s)),
    ops.filter(lambda i: i >= 5)
)

# Subscribe to the observable to start the data flow and print results
composed.subscribe(
    on_next=lambda value: print(f"Received {value}"),
    on_error=lambda err: print(f"Error: {err}"),
    on_completed=lambda: print("Done!")
)

view raw JSON →