{"id":2267,"library":"rx","title":"Reactive Extensions for Python (RxPY)","description":"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.","status":"active","version":"3.2.0","language":"en","source_language":"en","source_url":"https://github.com/ReactiveX/RxPY","tags":["reactive programming","async","streams","events","data flow"],"install":[{"cmd":"pip install rx","lang":"bash","label":"Install `rx` package (version 3.x)"}],"dependencies":[],"imports":[{"note":"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`.","wrong":"import reactivex","symbol":"rx","correct":"import rx"},{"note":"Operators are typically imported from the `rx.operators` submodule and aliased as `ops`.","symbol":"operators","correct":"from rx import operators as ops"}],"quickstart":{"code":"import rx\nfrom rx import operators as ops\n\n# Create an observable from a sequence of items\nsource = rx.of(\"Alpha\", \"Beta\", \"Gamma\", \"Delta\", \"Epsilon\")\n\n# Define a pipeline of operators\ncomposed = source.pipe(\n    ops.map(lambda s: len(s)),\n    ops.filter(lambda i: i >= 5)\n)\n\n# Subscribe to the observable to start the data flow and print results\ncomposed.subscribe(\n    on_next=lambda value: print(f\"Received {value}\"),\n    on_error=lambda err: print(f\"Error: {err}\"),\n    on_completed=lambda: print(\"Done!\")\n)\n","lang":"python","description":"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."},"warnings":[{"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`.","message":"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.","severity":"breaking","affected_versions":"3.x users migrating to 4.x+"},{"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.","message":"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`).","severity":"breaking","affected_versions":"3.x users migrating to 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`).","message":"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.","severity":"breaking","affected_versions":"3.x users migrating to 4.x+"},{"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))`).","message":"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.","severity":"gotcha","affected_versions":"All versions"}],"env_vars":null,"last_verified":"2026-04-09T00:00:00.000Z","next_check":"2026-07-08T00:00:00.000Z"}