223. Reactive Programming in Python

🔹 1. Installing RxPY

Copy

pip install rx

✅ Fix: Install the RxPY library before using it.

---

🔹 2. Simple Observable and Observer Example

Copy

import rx
from rx import operators as ops

def print_value(x):
    print(x)

observable = rx.from_([1, 2, 3, 4, 5])
observable.subscribe(print_value)

✅ Fix: rx.from_() creates an observable and subscribe() attaches observer to it.

---

🔹 3. Using map Operator to Transform Data

Copy

import rx
from rx import operators as ops

observable = rx.from_([1, 2, 3, 4, 5])
observable.pipe(
    ops.map(lambda x: x * 2)
).subscribe(lambda x: print(f"Transformed: {x}"))

✅ Fix: map() is used to transform items in the observable stream.

---

🔹 4. Using filter Operator to Filter Data

Copy

import rx
from rx import operators as ops

observable = rx.from_([1, 2, 3, 4, 5, 6])
observable.pipe(
    ops.filter(lambda x: x % 2 == 0)
).subscribe(lambda x: print(f"Even number: {x}"))

✅ Fix: filter() filters the data based on a condition.

---

🔹 5. Combining Multiple Observables with merge

Copy

import rx
from rx import operators as ops

observable1 = rx.from_([1, 2, 3])
observable2 = rx.from_([4, 5, 6])

rx.merge(observable1, observable2).subscribe(lambda x: print(f"Received: {x}"))

✅ Fix: merge() combines multiple observables into a single stream.

---

🔹 6. Handling Errors with on_error

Copy

import rx
from rx import operators as ops

def error_func(x):
    if x == 3:
        raise Exception("Error at 3!")
    return x

observable = rx.from_([1, 2, 3, 4, 5]).pipe(
    ops.map(error_func)
)

observable.subscribe(
    on_next=lambda x: print(f"Received: {x}"),
    on_error=lambda e: print(f"Error: {e}")
)

✅ Fix: on_error() allows handling errors in reactive streams.

---

🔹 7. Using concat for Sequential Execution

Copy

import rx
from rx import operators as ops

observable1 = rx.from_([1, 2])
observable2 = rx.from_([3, 4])

rx.concat(observable1, observable2).subscribe(lambda x: print(f"Received: {x}"))

✅ Fix: concat() ensures sequential execution of multiple observables.

---

🔹 8. Delayed Execution with interval

Copy

import rx
from rx import operators as ops
import time

observable = rx.interval(1.0)  # Emits every second
observable.subscribe(lambda x: print(f"Tick {x}"))
time.sleep(5)  # Sleep for 5 seconds to observe the emitted ticks

✅ Fix: interval() emits values at fixed time intervals.

---

🔹 9. Debouncing with debounce

Copy

import rx
from rx import operators as ops
import time

observable = rx.from_([1, 2, 3, 4, 5]).pipe(
    ops.debounce(lambda x: rx.timer(0.5))  # Wait 0.5 seconds after last event
)

observable.subscribe(lambda x: print(f"Received: {x}"))

✅ Fix: debounce() delays emissions until a certain time has passed.

---

🔹 10. Using zip to Combine Observables

Copy

import rx
from rx import operators as ops

observable1 = rx.from_([1, 2, 3])
observable2 = rx.from_([4, 5, 6])

rx.zip(observable1, observable2).subscribe(lambda x: print(f"Received: {x}"))

✅ Fix: zip() pairs up values from multiple observables.

---

🚀 Summary: Why Use RxPY?

FeatureRxPY

Reactive Programming

✅ Handle asynchronous events

Combining Observables

✅ merge, zip, concat

Transformations

✅ map, filter, debounce

Error Handling

✅ on_error

Scheduled Emissions

✅ interval, timer

---

🚀 When to Use RxPY?

ScenarioUse RxPY?

Handling real-time events (like UI updates)

✅ Yes

Managing complex asynchronous workflows

✅ Yes

Building event-driven systems or microservices

✅ Yes

Streaming data processing

✅ Yes

---