ReactiveX for Python v5 brings back method chaining support alongside the existing pipe-based functional style, giving developers the freedom to choose their preferred syntax:
Dual syntax support: Both fluent (method chaining) and functional (pipe) styles are now first-class citizens
Full type safety: Complete type inference in both styles with strict pyright and mypy checking
Backward compatible: All existing pipe-based code continues to work without any changes
Mix and match: Fluent and functional styles can be freely combined in the same operator chain
RxPY v5 brings back the intuitive method chaining syntax that many developers prefer. All 150+ operators are now available as methods on the Observable class:
import reactivex as rx
# Fluent style - chain operators as methods
(
rx.of("Alpha", "Beta", "Gamma", "Delta", "Epsilon")
.map(lambda s: len(s))
.filter(lambda i: i >= 5)
.subscribe(lambda value: print(f"Received {value}"))
)
This style is more concise, easier to read, and provides better IDE autocomplete support. Note the use of parentheses for implicit line continuation - no backslashes needed!
The pipe-based functional style from RxPY v4 continues to work exactly as before:
import reactivex as rx
from reactivex import operators as ops
# Functional style - compose operators with pipe
rx.of("Alpha", "Beta", "Gamma", "Delta", "Epsilon").pipe(
ops.map(lambda s: len(s)),
ops.filter(lambda i: i >= 5)
).subscribe(lambda value: print(f"Received {value}"))
Both styles produce identical results and have the same performance characteristics.
You can freely mix fluent and functional styles in the same chain:
import reactivex as rx
from reactivex import operators as ops
# Start with fluent style
(
rx.of(1, 2, 3, 4, 5)
.map(lambda x: x * 2)
.filter(lambda x: x > 5)
.pipe( # Switch to functional style
ops.take(2),
ops.reduce(lambda acc, x: acc + x),
)
.subscribe(print)
)
Choose the style that works best for your use case, or mix them as needed.
Here’s a complete example showing both styles:
import reactivex as rx
from reactivex import operators as ops
# Fluent style
source = rx.interval(0.1).take(5)
result = (
source.map(lambda x: x * 2)
.filter(lambda x: x > 3)
.scan(lambda acc, x: acc + x, 0)
.subscribe(lambda x: print(f"Fluent: {x}"))
)
# Functional style (equivalent)
source = rx.interval(0.1).take(5)
result = source.pipe(
ops.map(lambda x: x * 2),
ops.filter(lambda x: x > 3),
ops.scan(lambda acc, x: acc + x, 0)
).subscribe(lambda x: print(f"Functional: {x}"))
Both styles maintain full type safety with proper type inference:
from reactivex import Observable
# Type inference works in fluent style
source: Observable[int] = rx.of(1, 2, 3)
result: Observable[str] = source.map(str) # int -> str
# Type inference works in functional style
source2: Observable[int] = rx.of(1, 2, 3)
result2: Observable[str] = source2.pipe(ops.map(str))
The fluent style provides excellent IDE autocomplete support. When you type
source. your IDE will show all 150+ available operators with their
documentation and type signatures.
All existing RxPY v4 code continues to work without modification. The addition of method chaining is purely additive - no APIs were removed or changed.
Beyond the new fluent API, v5 modernizes the project itself. None of these affect application code:
Python support: Python 3.8 and 3.9 were dropped; v5 supports Python 3.10 through 3.14.
Project tooling: the build backend moved from Poetry to uv, and Black plus isort were replaced by Ruff for formatting and linting.
Operator fixes: several operators had scheduler-forwarding and
resubscription bugs corrected (for example timer resetting its delay on
each resubscription, and pairwise / to_marbles / delay_with_mapper
forwarding the scheduler argument). These are bug fixes, not API changes.
ReactiveX for Python v4 is an evolution of RxPY v3 to modernize it to current Python standards:
Project main module renamed from rx to reactivex. This is done
to give it a unique name different from the obsolete Reactive Extensions
(RxPY)
Generic type annotations. Code now type checks with pyright / pylance at strict settings. It also mostly type checks with mypy. Mypy should eventually catch up.
The pipe function has been renamed to compose. There is now a
new function pipe that works similar to the pipe method.
RxPY is now a modern Python project using pyproject.toml instead
of setup.py, and using modern tools such as Poetry, Black
formatter and isort. (As of v5 these have been replaced by uv and Ruff;
see Tooling and Python Support above.)
import reactivex as rx
from reactivex import operators as ops
rx.of("Alpha", "Beta", "Gamma", "Delta", "Epsilon").pipe(
ops.map(lambda s: len(s)),
ops.filter(lambda i: i >= 5)
).subscribe(lambda value: print("Received {0}".format(value)))
RxPY v3 is a major evolution from RxPY v1. This release brings many improvements, some of the most important ones being:
A better integration in IDEs via autocompletion support.
New operators can be implemented outside of RxPY.
Operator chains are now built via the pipe operator.
A default scheduler can be provided in an operator chain.
The most fundamental change is the way operators are chained together. On RxPY v1, operators were methods of the Observable class. So they were chained by using the existing Observable methods:
from rx import Observable
Observable.of("Alpha", "Beta", "Gamma", "Delta", "Epsilon") \
.map(lambda s: len(s)) \
.filter(lambda i: i >= 5) \
.subscribe(lambda value: print("Received {0}".format(value)))
Chaining in RxPY v3 is based on the pipe operator.
This operator is now one of the only methods of the
Observable class. In RxPY v3, operators are implemented
as functions:
import rx
from rx import operators as ops
rx.of("Alpha", "Beta", "Gamma", "Delta", "Epsilon").pipe(
ops.map(lambda s: len(s)),
ops.filter(lambda i: i >= 5)
).subscribe(lambda value: print("Received {0}".format(value)))
The fact that operators are functions means that adding new operators is now very easy. Instead of wrapping custom operators with the let operator, they can be directly used in a pipe chain.
The mapper function is removed in operators that combine the values of several
observables. This change applies to the following operators:
combine_latest,
group_join,
join,
with_latest_from,
zip, and
zip_with_iterable.
In RxPY v1, these operators were used the following way:
from rx import Observable
import operator
a = Observable.of(1, 2, 3, 4)
b = Observable.of(2, 2, 4, 4)
a.zip(b, lambda a, b: operator.mul(a, b)) \
.subscribe(print)
Now they return an Observable of tuples, with each item being the combination of the source Observables:
import rx
from rx 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.map(lambda z: operator.mul(z[0], z[1]))
).subscribe(print)
Dealing with the tuple unpacking is made easier with the starmap operator that unpacks the tuple to args:
import rx
from rx 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),
ops.starmap(operator.mul)
).subscribe(print)
The subscription function provided to the create operator
now takes two parameters: An observer and a scheduler. The scheduler parameter
is new: If a scheduler has been set in the call to subscribe, then this
scheduler is passed to the subscription function. Otherwise this parameter is
set to None.
One can use or ignore this parameter. This new scheduler parameter allows the create operator to use the default scheduler provided in the subscribe call. So scheduling item emissions with relative or absolute due-time is now possible.
The support of list of Observables as a parameter has been removed in the
following operators:
merge,
zip, and
combine_latest.
For example in RxPY v1 the merge operator could be called with a list:
from rx import Observable
obs1 = Observable.from_([1, 2, 3, 4])
obs2 = Observable.from_([5, 6, 7, 8])
res = Observable.merge([obs1, obs2])
res.subscribe(print)
This is not possible anymore in RxPY v3. So Observables must be provided explicitly:
import rx, operator as op
obs1 = rx.from_([1, 2, 3, 4])
obs2 = rx.from_([5, 6, 7, 8])
res = rx.merge(obs1, obs2)
res.subscribe(print)
If for any reason the Observables are only available as a list, then they can be unpacked:
import rx
from rx import operators as ops
obs1 = rx.from_([1, 2, 3, 4])
obs2 = rx.from_([5, 6, 7, 8])
obs_list = [obs1, obs2]
res = rx.merge(*obs_list)
res.subscribe(print)
BlockingObservables have been removed from rxPY v3. In RxPY v1, blocking until an Observable completes was done the following way:
from rx import Observable
res = Observable.from_([1, 2, 3, 4]).to_blocking().last()
print(res)
This is now done with the run operator:
import rx
res = rx.from_([1, 2, 3, 4]).run()
print(res)
The run operator returns only the last value emitted by the source Observable. It is possible to use the previous blocking operators by using the standard operators before run. For example:
Get first item: obs.pipe(ops.first()).run()
Get all items: obs.pipe(ops.to_list()).run()
Support for back-pressure - and so ControllableObservable - has been removed in RxPY v3. Back-pressure can be implemented in several ways, and many strategies can be adopted. So we consider that such features are beyond the scope of RxPY. You are encouraged to provide independent implementations as separate packages so that they can be shared by the community.
List of community projects supporting backpressure can be found in Additional Reading.
Operators that take time values as parameters now use seconds as a unit instead of milliseconds. This RxPY v1 example:
ops.debounce(500)
is now written as:
ops.debounce(0.5)
Some packages were renamed:
Old name |
New name |
rx.concurrency |
reactivex.scheduler |
rx.disposables |
rx.disposable |
rx.subjects |
rx.subject |
Furthermore, the package formerly known as rx.concurrency.mainloopscheduler has been split into two parts, reactivex.scheduler.mainloop and reactivex.scheduler.eventloop.