ReactiveX for Python (RxPY)

Warnings

• breaking Major API changes occurred during the migration from RxPY v3 to v4. The main module name changed from `rx` to `reactivex`. Operator chaining shifted from method-based calls (e.g., `Observable.map().filter()`) to a functional `pipe` method that takes operators from `reactivex.operators` (e.g., `Observable.pipe(ops.map(), ops.filter())`). The `pipe` function itself was renamed to `compose`, with a new `pipe` method for chaining.
  Fix: Update imports from `rx` to `reactivex`. Rewrite operator chains to use the `pipe` method and import operators from `reactivex.operators`. Refer to the migration guide for specific operator changes.
• breaking Python 3.10+ deprecations introduced breaking changes in RxPY v4 related to concurrency primitives. Specifically, `setDaemon` (deprecated in Python 3.10) was replaced, and the `loop` parameter for methods like `create` and other concurrency-related functions was removed. Coroutine decorators were replaced with the `async` keyword. Users interacting directly with event loops or threading might need adjustments.
  Fix: Review code for direct usage of Python's deprecated `threading.setDaemon` or explicit `loop` parameters in RxPY's `create` or scheduler functions. Adopt modern `async/await` syntax where applicable. For custom schedulers or advanced concurrency, consult the RxPY v4 documentation on schedulers.
• gotcha In Python 3.10 and above, `asyncio.get_event_loop()` is deprecated, and `asyncio.get_running_loop()` or `asyncio.new_event_loop()` combined with `asyncio.set_event_loop()` should be used instead. While RxPY v5 alpha has internally addressed this, user code that directly interacts with `asyncio` event loops for custom schedulers or integration might encounter `DeprecationWarning`s or `RuntimeError`s on newer Python versions.
  Fix: For explicit `asyncio` event loop management, use `asyncio.get_running_loop()` within an `async` context, or `asyncio.new_event_loop()` and `asyncio.set_event_loop()` for setting a new loop. Consider using RxPY's provided schedulers which are designed to handle these Python version differences internally.
• gotcha While RxPY v5 introduces method chaining (fluent style) as an additive feature alongside the existing pipe-based functional style, some users might find it confusing to switch between the two. Both styles are fully type-safe and can be mixed.
  Fix: Choose a consistent style (fluent or functional) for clarity within your project, or leverage both where it enhances readability. Be aware that most v4 examples will be pipe-based, and v5 will offer both.

Install

• pip install reactivex Install stable version
• pip install 'reactivex>=5.0.0a0' Install pre-release v5.x (alpha)

Imports

• reactivex

  import reactivex as rx

  The main module was renamed from `rx` to `reactivex` in v4 to avoid naming conflicts and align with modern Python practices.
• operators

  from reactivex import operators as ops

  Operators are typically imported from the `reactivex.operators` submodule and aliased as `ops` for conciseness in chaining.

Quickstart

This quickstart demonstrates creating an observable from a sequence of strings, transforming them using `map` (to get length) and `filter` (to keep lengths >= 5), and then subscribing to process the results. It highlights the functional `pipe` operator for chaining transformations and the use of `on_next`, `on_error`, and `on_completed` handlers.

import reactivex as rx
from reactivex import operators as ops

def my_on_next(value):
    print(f"Received: {value}")

def my_on_error(error):
    print(f"Error: {error}")

def my_on_completed():
    print("Done!")

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

disposable = source.pipe(
    ops.map(lambda s: len(s)),
    ops.filter(lambda i: i >= 5)
).subscribe(
    on_next=my_on_next,
    on_error=my_on_error,
    on_completed=my_on_completed
)

# For long-running observables or specific cleanup, dispose is important
disposable.dispose()