ReactiveX for Python (RxPY)

ReactiveX for Python (RxPY) is used to compose asynchronous and event-based programs. It relies on three primary concepts:

1. Observables: Represent asynchronous data streams (e.g., stock quotes, Tweets, computer events, or web service requests).
2. Observers: Objects that subscribe to an Observable and are notified whenever an event occurs in the stream.
3. Operators: Functions used to query and transform data streams. These can be chained using the pipe operator to perform operations like filter, map, reduce, and time-based transformations.
4. Schedulers: Used to parameterize concurrency within data and event streams.

For foundational knowledge, operator details, and guided tutorials, refer to the official ReactiveX resources:

• Introduction: General overview of ReactiveX.
• Tutorials: Step-by-step learning paths.
• Operators: Detailed documentation of available operators.

http://reactivex.io/intro.html
http://reactivex.io/tutorials.html
http://reactivex.io/documentation/operators.html

When migrating from v3 to v4:

• The main module was renamed from rx to reactivex.
• Generic type annotations were added for better compatibility with pyright and mypy.
• The pipe function was renamed to compose, and a new pipe method was introduced to work similarly to the pipe method on Observables.

In RxPY v3, operators that combine multiple Observables (like zip, combine_latest, join, etc.) used a result mapper function. In v4+, these operators no longer accept a mapper; instead, they return an Observable of tuples. To process these tuples, use ops.map or ops.starmap.

import reactivex as rx
from reactivex import operators as ops
import operator

a = rx.of(1, 2, 3, 4)
b = rx.of(2, 2, 4, 4)

a.pipe(
    ops.zip(b), # returns a tuple with the items of a and b
    ops.starmap(operator.mul)
).subscribe(print)

RxPY v4.x only supported the functional (pipe-based) style. Because RxPY v5.x maintains full backward compatibility, all existing v4.x code using .pipe() and reactivex.operators will continue to work without any changes. There are no breaking changes for functional-style code.

# RxPY v4.x code works unchanged in v5.x
from reactivex import operators as ops

result = source.pipe(
    ops.map(lambda x: x * 2),
    ops.filter(lambda x: x > 5)
)

Instead of providing a scheduler to every individual operator, you can specify a default scheduler for the entire chain by passing it to the .subscribe() method.

Operators will select a scheduler in this priority order:

1. The scheduler explicitly provided to the operator.
2. The default scheduler provided in the subscribe call.
3. The operator's own internal default scheduler.

RxPY supports two styles for chaining operators to transform emissions:

1. Functional style: Uses the .pipe() method to pass operators (imported from reactivex.operators). This is often preferred for readability in complex pipelines.
2. Fluent style: Uses method chaining directly on the Observable object (e.g., .map().filter()).

Note: In v4, rx.pipe was renamed to compose for internal composition, but the .pipe() method on Observables remains the standard for building pipelines.

# Functional style
from reactivex import of, operators as op

source = of("Alpha", "Beta", "Gamma", "Delta", "Epsilon")
composed = source.pipe(
    op.map(lambda s: len(s)),
    op.filter(lambda i: i >= 5)
)

# Fluent style
composed = source.map(lambda s: len(s)).filter(lambda i: i >= 5)

In RxPY v3, merge, zip, and combine_latest accepted a list of Observables. In v4+, you must provide Observables as individual arguments. If you have a list, unpack it using the * operator.

import reactivex as rx
from reactivex import operators as ops

obs1 = rx.from_([1, 2, 3, 4])
obs2 = rx.from_([5, 6, 7, 8])

obs_list = [obs1, obs2]

# Unpack the list to pass as individual arguments
res = rx.merge(*obs_list)

To use RxPY v5, ensure you are running Python 3.10 or above. You can install the package using pip3.

pip3 install reactivex

The BlockingObservables API was removed in v3. To block until an Observable completes and get a result, use the .run() operator.

• To get the last value: obs.run()
• To get the first item: obs.pipe(ops.first()).run()
• To get all items as a list: obs.pipe(ops.to_list()).run()

import reactivex as rx

res = rx.from_([1, 2, 3, 4]).run()
print(res)