### Install Pyventus with All Integrations

Source: https://github.com/mdapena/pyventus/blob/master/docs/getting-started.md

Installs Pyventus along with all available optional dependencies, including support for Celery, Redis Queue, and FastAPI, providing maximum flexibility for processing services.

```console
pip install pyventus[all]
```

--------------------------------

### Install Pyventus with Celery Support

Source: https://github.com/mdapena/pyventus/blob/master/docs/getting-started.md

Installs Pyventus along with the necessary dependencies to integrate with the Celery framework for asynchronous task processing. This enables the use of CeleryProcessingService.

```console
pip install pyventus[celery]
```

--------------------------------

### Install Pyventus using pip

Source: https://github.com/mdapena/pyventus/blob/master/README.md

Installs the Pyventus Python package using pip. This is the recommended way to get started with Pyventus, ensuring proper dependency management.

```console
pip install pyventus
```

--------------------------------

### Install Pyventus with Redis Queue (RQ) Support

Source: https://github.com/mdapena/pyventus/blob/master/docs/getting-started.md

Installs Pyventus with optional dependencies for Redis Queue (RQ) integration. This allows Pyventus to leverage RQ for managing and executing tasks via Redis.

```console
pip install pyventus[rq]
```

--------------------------------

### Install Pyventus with FastAPI Support

Source: https://github.com/mdapena/pyventus/blob/master/docs/getting-started.md

Installs Pyventus with dependencies required for FastAPI integration. This enables the use of FastAPI's BackgroundTasks for handling event processing within FastAPI applications.

```console
pip install pyventus[fastapi]
```

--------------------------------

### Pyventus Basic Event Handling Example

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Demonstrates basic event-driven programming in Pyventus. It shows how to link an event ('GreetEvent') to a callback function and emit the event using an EventEmitter.

```python
from pyventus.events import AsyncIOEventEmitter, EventEmitter, EventLinker


@EventLinker.on("GreetEvent")
def handle_greet_event():
    print("Hello, World!")


event_emitter: EventEmitter = AsyncIOEventEmitter()
event_emitter.emit("GreetEvent")
```

--------------------------------

### Install Pyventus

Source: https://github.com/mdapena/pyventus/blob/master/docs/getting-started.md

Installs the core Pyventus package. It relies on the Python standard library and requires Python 3.10 or higher. The typing-extensions package is also a dependency for advanced typing features.

```console
pip install pyventus
```

--------------------------------

### Install Pyventus with Pip

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Installs the Pyventus Python package using pip. It's recommended to install within a virtual environment for dependency isolation. Requires Python 3.10+.

```console
pip install pyventus
```

--------------------------------

### Pyventus Async Event Handling Example

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Illustrates asynchronous event handling in Pyventus. This example uses an async function as a callback for an event, demonstrating compatibility with async contexts.

```python
from pyventus.events import AsyncIOEventEmitter, EventEmitter, EventLinker


@EventLinker.on("GreetEvent")
async def handle_greet_event():
    print("Hello, World!")


event_emitter: EventEmitter = AsyncIOEventEmitter()
event_emitter.emit("GreetEvent")
```

--------------------------------

### Simple Counter Example (Python)

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Demonstrates defining a synchronous observable task using the `@as_observable_task` decorator. It yields numbers sequentially and signals completion. The example shows how to subscribe to the observable and print received values.

```Python
from pyventus.reactive import as_observable_task, Completed


@as_observable_task
def simple_counter(stop: int):
    for count in range(1, stop + 1):
        yield count
    raise Completed


obs = simple_counter(stop=16)
obs.subscribe(
    next_callback=lambda val: print(f"Received: {val}"),
    complete_callback=lambda: print("Done!"),
)
obs()
```

--------------------------------

### Simple Counter Example (Async Python)

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Illustrates defining an asynchronous observable task using the `@as_observable_task` decorator. This version handles async operations, yielding numbers and signaling completion. It demonstrates subscribing to the async observable and printing received values.

```Python
from pyventus.reactive import as_observable_task, Completed


@as_observable_task
async def simple_counter(stop: int):
    for count in range(1, stop + 1):
        yield count
    raise Completed


obs = simple_counter(stop=16)
obs.subscribe(
    next_callback=lambda val: print(f"Received: {val}"),
    complete_callback=lambda: print("All done!"),
)
obs()
```

--------------------------------

### Pyventus: Basic Event Handling Example

Source: https://github.com/mdapena/pyventus/blob/master/README.md

Demonstrates basic event-driven programming with Pyventus. It shows how to link an event ('GreetEvent') to a callback function and emit the event using an EventEmitter.

```python
from pyventus.events import AsyncIOEventEmitter, EventEmitter, EventLinker


@EventLinker.on("GreetEvent")
def handle_greet_event():
    print("Hello, World!")


event_emitter: EventEmitter = AsyncIOEventEmitter()
event_emitter.emit("GreetEvent")
```

--------------------------------

### Simple Counter Example (Python)

Source: https://github.com/mdapena/pyventus/blob/master/README.md

Demonstrates defining a synchronous observable task using the `@as_observable_task` decorator. It yields numbers sequentially and signals completion. The example shows how to subscribe to the observable and print received values.

```Python
from pyventus.reactive import as_observable_task, Completed


@as_observable_task
def simple_counter(stop: int):
    for count in range(1, stop + 1):
        yield count
    raise Completed


obs = simple_counter(stop=16)
obs.subscribe(
    next_callback=lambda val: print(f"Received: {val}"),
    complete_callback=lambda: print("Done!"),
)
obs()
```

--------------------------------

### Pyventus: Async Event Handling Example

Source: https://github.com/mdapena/pyventus/blob/master/README.md

Illustrates asynchronous event handling in Pyventus. This example uses `AsyncIOEventEmitter` and an `async` callback function to handle events, highlighting support for asynchronous operations.

```python
from pyventus.events import AsyncIOEventEmitter, EventEmitter, EventLinker


@EventLinker.on("GreetEvent")
async def handle_greet_event():
    print("Hello, World!")


event_emitter: EventEmitter = AsyncIOEventEmitter()
event_emitter.emit("GreetEvent")
```

--------------------------------

### Install Development Dependencies

Source: https://github.com/mdapena/pyventus/blob/master/docs/contributing.md

Installs the project's development dependencies, including testing and linting tools, using pip. The `.[dev]` extra installs development-specific packages.

```console
pip install -e .[dev]
```

--------------------------------

### Declarative Subscription with Context Manager

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Demonstrates using the `subscribe()` method as a context manager to define observer callbacks inline. This approach allows for defining `on_next`, `on_error`, and `on_complete` handlers within a `with` block.

```Python
with obs.subscribe() as subctx:

    @subctx.on_next
    def next(value: int) -> None:
        print(f"Received: {value}")

    @subctx.on_error
    def error(error: Exception) -> None:
        print(f"Error: {error}")

    @subctx.on_complete
    def complete() -> None:
        print("All done!")
```

--------------------------------

### Install Pyventus Benchmark Dependencies

Source: https://github.com/mdapena/pyventus/blob/master/docs/release-notes.md

Installs required packages for Pyventus benchmarks. Includes `pyventus[tests]` for testing capabilities and `matplotlib` for result visualization. These commands ensure the environment is ready for benchmark execution.

```Shell
pip install pyventus[tests]
pip install matplotlib
```

--------------------------------

### Pyventus: Subscribe to Events with Callbacks via subscribe() Method

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Provides an alternative method for setting up event callbacks in Pyventus using the `subscribe()` function. This approach is suitable for simpler definitions like lambda functions or when using pre-defined functions, allowing direct specification of event, success, and failure callbacks.

```Python
from pyventus.events import AsyncIOEventEmitter, EventEmitter, EventLinker

EventLinker.subscribe(
    "DivisionEvent",
    event_callback=lambda a, b: a / b,
    success_callback=lambda result: print(f"Division result: {result:.3g}"),
    failure_callback=lambda e: print(f"Oops, something went wrong: {e}"),
)

event_emitter: EventEmitter = AsyncIOEventEmitter()
event_emitter.emit("DivisionEvent", a=1, b=0)
```

--------------------------------

### Celery Worker Setup

Source: https://github.com/mdapena/pyventus/blob/master/docs/learn/events/emitters/celery.md

Example setup for a Celery worker, including initializing the Celery app with a broker (e.g., Redis) and registering the Pyventus shared task.

```Python
from celery import Celery
from pyventus.core.processing.celery import CeleryProcessingService

# Initialize a Celery app with Redis as an example broker; other brokers are also supported.
celery_app: Celery = Celery(broker="redis://default:redispw@localhost:6379")

# Register the Pyventus shared task.
CeleryProcessingService.register()

# Start the Celery worker.
if __name__ == "__main__":
    celery_app.start()
```

--------------------------------

### Simplified Observable Task Execution Context

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Illustrates how to use observable tasks within a `with` statement block. This enables an execution context that automatically calls the observable tasks upon exiting the block, simplifying manual invocation.

```Python
@as_observable_task
def simple_counter(stop: int):
    for count in range(1, stop + 1):
        yield count
    raise Completed

with simple_counter(stop=16) as obs:
    obs.subscribe(lambda val: print(f"Subscriber 1 - Received: {val}"))
    obs.subscribe(lambda val: print(f"Subscriber 2 - Received: {val}"))
```

--------------------------------

### Pyventus: Handle Event Success and Failure with Context Managers

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Illustrates custom success and error handling for events in Pyventus using a context manager (`with EventLinker.on(...)`). It defines separate callbacks for successful event execution (`on_success`) and for exceptions (`on_failure`), demonstrating how to manage outcomes for specific events.

```Python
from pyventus.events import AsyncIOEventEmitter, EventEmitter, EventLinker

with EventLinker.on("DivisionEvent") as subctx:

    @subctx.on_event
    def divide(a: float, b: float) -> float:
        return a / b

    @subctx.on_success
    def handle_success(result: float) -> None:
        print(f"Division result: {result:.3g}")

    @subctx.on_failure
    def handle_failure(e: Exception) -> None:
        print(f"Oops, something went wrong: {e}")


event_emitter: EventEmitter = AsyncIOEventEmitter()
event_emitter.emit("DivisionEvent", a=1, b=0)
event_emitter.emit("DivisionEvent", a=1, b=2)
```

--------------------------------

### Dynamic Voltage Monitoring (Event-Driven Implementation)

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Demonstrates an event-driven approach to monitoring voltage sensors using Pyventus. It defines a `VoltageSensor` class that simulates readings and emits `LowVoltageEvent` or `HighVoltageEvent` based on predefined thresholds. Event handlers are registered to react to these events, showcasing how to attach domain logic without modifying the sensor implementation.

```Python
import asyncio
import random

from pyventus.events import AsyncIOEventEmitter, EventEmitter, EventLinker


class VoltageSensor:

    def __init__(self, name: str, low: float, high: float, event_emitter: EventEmitter) -> None:
        # Initialize the VoltageSensor object with the provided parameters
        self._name: str = name
        self._low: float = low
        self._high: float = high
        self._event_emitter: EventEmitter = event_emitter

    async def __call__(self) -> None:
        # Start voltage readings for the sensor
        print(f"Starting voltage readings for: {self._name}")
        print(f"Low: {self._low:.3g}v | High: {self._high:.3g}v\n-----------")

        while True:
            # Simulate sensor readings
            voltage: float = random.uniform(0, 5)
            print("\tSensor Reading:", "\033[32m", f"{voltage:.3g}v", "\033[0m")

            # Emit events based on voltage readings
            if voltage < self._low:
                self._event_emitter.emit("LowVoltageEvent", sensor=self._name, voltage=voltage)
            elif voltage > self._high:
                self._event_emitter.emit("HighVoltageEvent", sensor=self._name, voltage=voltage)

            await asyncio.sleep(1)


@EventLinker.on("LowVoltageEvent")
def handle_low_voltage_event(sensor: str, voltage: float):
    print(f"🪫 Starting low-voltage protection for '{sensor}'. ({voltage:.3g}v)\n")
    # Perform action for low voltage...


@EventLinker.on("HighVoltageEvent")
def handle_high_voltage_event(sensor: str, voltage: float):
    print(f"⚡ Starting high-voltage protection for '{sensor}'. ({voltage:.3g}v)\n")
    # Perform action for high voltage...


@EventLinker.on("LowVoltageEvent", "HighVoltageEvent")
async def handle_voltage_event(sensor: str, voltage: float):
    print(f"\033[31m\nSensor '{sensor}' out of range.\033[0m (Voltage: {voltage:.3g})")
    # Perform notification for out of range voltage...


async def main():
    # Initialize the sensor and run the sensor readings
    sensor = VoltageSensor(name="CarSensor", low=0.5, high=3.9, event_emitter=AsyncIOEventEmitter())
    await asyncio.gather(sensor(), )  # Add new sensors inside the 'gather' for multi-device monitoring


asyncio.run(main())

```

--------------------------------

### Thread Offloading for Observable Tasks

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Demonstrates running observable tasks in separate threads using `concurrent.futures.ThreadPoolExecutor`. This is useful for multithreaded environments where default AsyncIO processing is not suitable.

```Python
from concurrent.futures import ThreadPoolExecutor
from pyventus.reactive import as_observable_task, Completed

@as_observable_task
def simple_counter(stop: int):
    for count in range(1, stop + 1):
        yield count
    raise Completed

if __name__ == "__main__":

    with ThreadPoolExecutor() as executor:

        obs1 = simple_counter(16)
        obs1.subscribe(print)
        obs1(executor)

        obs2 = simple_counter(16)
        obs2.subscribe(print)
        obs2(executor)
```

--------------------------------

### Create AsyncIO Event Emitter using Factory Method (Python)

Source: https://github.com/mdapena/pyventus/blob/master/docs/learn/events/emitters/asyncio.md

Shows how to create an AsyncIOEventEmitter using a factory method. This simplifies the setup process by handling the instantiation of the AsyncIOProcessingService internally, providing a more concise way to get started.

```Python
from pyventus.events import AsyncIOEventEmitter, EventEmitter


event_emitter: EventEmitter = AsyncIOEventEmitter()
event_emitter.emit("MyEvent")
```

--------------------------------

### Thread Offloading with Observable Task Execution Context

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Combines thread offloading with the execution context for observable tasks. Tasks are run in separate threads and automatically executed when their respective `with` blocks are exited.

```Python
from concurrent.futures import ThreadPoolExecutor
from pyventus.reactive import as_observable_task, Completed

@as_observable_task
def simple_counter(stop: int):
    for count in range(1, stop + 1):
        yield count
    raise Completed

if __name__ == "__main__":

    with ThreadPoolExecutor() as executor:

        with simple_counter(16).to_thread(executor) as obs1:
            obs1.subscribe(print)

        with simple_counter(16).to_thread(executor) as obs2:
            obs2.subscribe(print)
```

--------------------------------

### Pyventus Benchmark Script Structure

Source: https://github.com/mdapena/pyventus/blob/master/docs/release-notes.md

Illustrates the main execution block for running Pyventus benchmarks, including the setup and execution of the EventEmitterBenchmark.

```python
if __name__ == "__main__":
    main()
```

--------------------------------

### Declarative Subscription with Decorator

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Shows how to use the `subscribe()` method as a decorator. When used this way, it automatically creates and subscribes an observer, using the decorated function as its `on_next` callback.

```Python
@obs.subscribe()
def next(value: int) -> None:
    print(f"Received: {value}")
```

--------------------------------

### Benchmark Report Example (JSON)

Source: https://github.com/mdapena/pyventus/blob/master/docs/release-notes.md

An example JSON structure for a benchmark report, detailing the configuration used and the measurements collected for event emission time.

```json
{
  "title": "EventEmitterBenchmark(event_subscription_mode=SINGLE, onetime_subscription_mode=NONE)",
  "pyventus_version": "0.6.0",
  "benchmark_duration": 1555.6348432,
  "event_subscription_mode": "Single",
  "onetime_subscription_mode": "None",
  "subscription_sizes": [
    100,
    500,
    1000,
    5000,
    10000
  ],
  "num_repeats": 5,
  "num_executions": 1250,
  "measurements": [
    {
      "execution_time": 0.00123456789
    }
  ]
}
```

--------------------------------

### Pyventus: Subscribe to All Events with Global Event Listeners

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Shows how to use global event listeners in Pyventus by subscribing to a wildcard pattern (...). This is useful for implementing cross-cutting concerns like logging or monitoring, as it captures all events emitted within a specific EventLinker context.

```Python
@EventLinker.on(...)
def logging(*args, **kwargs):
    print(f"Logging:\n- Args: {args}\n- Kwargs: {kwargs}")


event_emitter: EventEmitter = AsyncIOEventEmitter()
event_emitter.emit("AnyEvent", name="Pyventus")
```

--------------------------------

### Customizable Success and Error Handling

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Customize how data streams are handled upon completion or error by configuring complete and error callbacks during the subscription process. This allows for tailored responses to stream events.

```Python
@as_observable_task
async def interactive_counter():
    stop: int = int(input("Please enter a number to count up to: "))  # Can raise ValueError
    for count in range(1, stop + 1):
        yield count
    raise Completed


obs = interactive_counter()
obs.subscribe(
    next_callback=lambda val: print(f"Received: {val}"),
    error_callback=lambda err: print(f"Error: {err}"),
    complete_callback=lambda: print("All done!"),
)
obs()
```

--------------------------------

### Runtime Event Emitter Flexibility

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Pyventus's modular design allows changing the event emitter at runtime without reconfiguring subscriptions. This is demonstrated by switching between `AsyncIOEventEmitter` and `ExecutorEventEmitter`.

```Python
from concurrent.futures import ThreadPoolExecutor

from pyventus.events import AsyncIOEventEmitter, EventEmitter, EventLinker, ExecutorEventEmitter


@EventLinker.on("Event")
def handle_event(msg: str):
    print(msg)


def main(event_emitter: EventEmitter) -> None:
    event_emitter.emit("Event", msg=f"{event_emitter}")


if __name__ == "__main__":
    executor = ThreadPoolExecutor()
    main(event_emitter=AsyncIOEventEmitter())
    main(event_emitter=ExecutorEventEmitter(executor))
    executor.shutdown()
```

--------------------------------

### Define Sync and Async Event Callbacks

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Pyventus allows defining event callbacks as either synchronous or asynchronous functions. This enables seamless integration into both sync and async Python applications.

```Python
@EventLinker.on("MyEvent")
def sync_event_callback():
    pass  # Synchronous event handling


@EventLinker.on("MyEvent")
async def async_event_callback():
    pass  # Asynchronous event handling
```

--------------------------------

### Simple Counter Example (Async Python)

Source: https://github.com/mdapena/pyventus/blob/master/README.md

Illustrates defining an asynchronous observable task using the `@as_observable_task` decorator. This version handles async operations, yielding numbers and signaling completion. It demonstrates subscribing to the async observable and printing received values.

```Python
from pyventus.reactive import as_observable_task, Completed


@as_observable_task
async def simple_counter(stop: int):
    for count in range(1, stop + 1):
        yield count
    raise Completed


obs = simple_counter(stop=16)
obs.subscribe(
    next_callback=lambda val: print(f"Received: {val}"),
    complete_callback=lambda: print("All done!"),
)
obs()
```

--------------------------------

### FastAPI: Server Output Example (Console)

Source: https://github.com/mdapena/pyventus/blob/master/docs/learn/events/emitters/fastapi.md

Example console output from running a FastAPI application with Pyventus. It shows server startup logs, incoming requests, and the execution of event handlers, including simulated delays.

```console
INFO   Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO   Started reloader process [12604] using WatchFiles
INFO   Started server process [7008]
INFO   Waiting for application startup.
INFO   Application startup complete.
INFO   127.0.0.1:49763 - "GET /email?email_to=email@email.com HTTP/1.1" 200
Sending email to: email@email.com
Email sent successfully!
```

--------------------------------

### Configure AsyncIO Event Emitter with Manual Setup (Python)

Source: https://github.com/mdapena/pyventus/blob/master/docs/learn/events/emitters/asyncio.md

Demonstrates manual configuration of an EventEmitter with the AsyncIOProcessingService. This approach explicitly creates and passes the service instance to the EventEmitter, allowing for direct control over the event processing setup.

```Python
from pyventus.events import EventEmitter
from pyventus.core.processing.asyncio import AsyncIOProcessingService

event_emitter = EventEmitter(event_processor=AsyncIOProcessingService())
event_emitter.emit("MyEvent")
```

--------------------------------

### Manual EventEmitter Setup with ExecutorProcessingService

Source: https://github.com/mdapena/pyventus/blob/master/docs/learn/events/emitters/executor.md

This snippet shows how to manually set up an `EventEmitter` by creating an `ExecutorProcessingService` instance with a `ThreadPoolExecutor` and passing it to the `EventEmitter` constructor. It demonstrates the core integration pattern for thread-based event processing.

```Python
from concurrent.futures import ThreadPoolExecutor

from pyventus.core.processing.executor import ExecutorProcessingService
from pyventus.events import EventEmitter


if __name__ == "__main__":
    executor = ThreadPoolExecutor()

    event_emitter = EventEmitter(event_processor=ExecutorProcessingService(executor))
    event_emitter.emit("MyEvent")

    executor.shutdown()
```

--------------------------------

### FastAPI: Start Server Command (Console)

Source: https://github.com/mdapena/pyventus/blob/master/docs/learn/events/emitters/fastapi.md

Command to start the FastAPI development server using uvicorn. This command is used to run the application and test the event emission functionality.

```console
fastapi dev main.py
```

--------------------------------

### Declarative Subscription with Context Manager

Source: https://github.com/mdapena/pyventus/blob/master/README.md

Demonstrates using the `subscribe()` method as a context manager to define observer callbacks inline. This approach allows for defining `on_next`, `on_error`, and `on_complete` handlers within a `with` block.

```Python
with obs.subscribe() as subctx:

    @subctx.on_next
    def next(value: int) -> None:
        print(f"Received: {value}")

    @subctx.on_error
    def error(error: Exception) -> None:
        print(f"Error: {error}")

    @subctx.on_complete
    def complete() -> None:
        print("All done!")
```

--------------------------------

### Manual Configuration of Redis Event Emitter

Source: https://github.com/mdapena/pyventus/blob/master/docs/learn/events/emitters/redis.md

Demonstrates manual setup of an EventEmitter with RedisProcessingService. It requires creating a Redis connection, a default RQ queue, and then initializing the EventEmitter with the RedisProcessingService instance. Events are emitted using the `emit` method.

```Python
from pyventus.core.processing.redis import RedisProcessingService
from pyventus.events import EventEmitter
from redis import Redis
from rq import Queue

redis_conn = Redis.from_url("redis://default:redispw@localhost:6379")
default_queue: Queue = Queue(name="default", connection=redis_conn)

if __name__ == "__main__":
    event_emitter: EventEmitter = EventEmitter(event_processor=RedisProcessingService(queue=default_queue))
    event_emitter.emit("MyEvent")
```

--------------------------------

### Event Emitter Utilities

Source: https://github.com/mdapena/pyventus/blob/master/docs/release-notes.md

Provides utilities for creating preconfigured event emitter instances, simplifying setup and ensuring retro compatibility with previous class-based designs.

```APIDOC
Event Emitter Utilities:
  Utilities for creating preconfigured event emitter instances.
  Simplifies setup and maintains backward compatibility.
```

--------------------------------

### Pyventus: Emit Structured Events with Event Objects

Source: https://github.com/mdapena/pyventus/blob/master/docs/index.md

Demonstrates how to define and emit custom event objects in Pyventus using Python dataclasses. Event objects provide a structured way to encapsulate event data payloads. Subscribers can be attached directly to these event classes.

```Python
@dataclass
class OrderCreatedEvent:
    order_id: int
    payload: dict


@EventLinker.on(OrderCreatedEvent)
def handle_order_created_event(event: OrderCreatedEvent):
    print(f"Event Object: {event}")

event_emitter: EventEmitter = AsyncIOEventEmitter()
event_emitter.emit(
    event=OrderCreatedEvent(
        order_id=6452879,
        payload={},
    ),
)
```

--------------------------------

### Declarative Subscription with Decorator

Source: https://github.com/mdapena/pyventus/blob/master/README.md

Shows how to use the `subscribe()` method as a decorator. When used this way, it automatically creates and subscribes an observer, using the decorated function as its `on_next` callback.

```Python
@obs.subscribe()
def next(value: int) -> None:
    print(f"Received: {value}")
```

--------------------------------

### Pyventus: Handle Event Success and Failure with Context Managers

Source: https://github.com/mdapena/pyventus/blob/master/README.md

Illustrates custom success and error handling for events in Pyventus using a context manager (`with EventLinker.on(...)`). It defines separate callbacks for successful event execution (`on_success`) and for exceptions (`on_failure`), demonstrating how to manage outcomes for specific events.

```Python
from pyventus.events import AsyncIOEventEmitter, EventEmitter, EventLinker

with EventLinker.on("DivisionEvent") as subctx:

    @subctx.on_event
    def divide(a: float, b: float) -> float:
        return a / b

    @subctx.on_success
    def handle_success(result: float) -> None:
        print(f"Division result: {result:.3g}")

    @subctx.on_failure
    def handle_failure(e: Exception) -> None:
        print(f"Oops, something went wrong: {e}")


event_emitter: EventEmitter = AsyncIOEventEmitter()
event_emitter.emit("DivisionEvent", a=1, b=0)
event_emitter.emit("DivisionEvent", a=1, b=2)
```

--------------------------------

### Pyventus: Subscribe to Events with Callbacks via subscribe() Method

Source: https://github.com/mdapena/pyventus/blob/master/README.md

Provides an alternative method for setting up event callbacks in Pyventus using the `subscribe()` function. This approach is suitable for simpler definitions like lambda functions or when using pre-defined functions, allowing direct specification of event, success, and failure callbacks.

```Python
from pyventus.events import AsyncIOEventEmitter, EventEmitter, EventLinker

EventLinker.subscribe(
    "DivisionEvent",
    event_callback=lambda a, b: a / b,
    success_callback=lambda result: print(f"Division result: {result:.3g}"),
    failure_callback=lambda e: print(f"Oops, something went wrong: {e}"),
)

event_emitter: EventEmitter = AsyncIOEventEmitter()
event_emitter.emit("DivisionEvent", a=1, b=0)
```

--------------------------------

### FastAPI: Practical Event Emission Example (Python)

Source: https://github.com/mdapena/pyventus/blob/master/docs/learn/events/emitters/fastapi.md

Illustrates a practical scenario where an event ('SendEmail') is emitted from a FastAPI endpoint. It includes defining an event handler using EventLinker and shows how to trigger the event with parameters.

```Python
import time

from fastapi import Depends, FastAPI
from pyventus.events import EventEmitter, EventLinker, FastAPIEventEmitter


@EventLinker.on("SendEmail")
def handle_email_notification(email: str) -> None:
	print(f"Sending email to: {email}")
	time.sleep(2.5)  # Simulate sending delay.
	print("Email sent successfully!")


app = FastAPI()


@app.get("/email")
def send_email(
	email_to: str,
	event_emitter: EventEmitter = Depends(FastAPIEventEmitter()),
) -> None:
	event_emitter.emit("SendEmail", email_to)
	return {"message": "Email sent!"}
```

--------------------------------

### Practical Example: ThreadPoolExecutor with Pyventus

Source: https://github.com/mdapena/pyventus/blob/master/docs/learn/events/emitters/executor.md

A practical example showcasing event emission using `ThreadPoolExecutor`. It demonstrates subscribing multiple callbacks to the same event and emitting the event twice, illustrating concurrent execution of event handlers.

```Python
import time
from concurrent.futures import ThreadPoolExecutor

from pyventus.events import EventLinker, ExecutorEventEmitter


def handle_event_with_delay():
    time.sleep(1.5)
    print("Done!")


if __name__ == "__main__":
    print("Starting...")

    EventLinker.subscribe("MyEvent", event_callback=handle_event_with_delay)
    EventLinker.subscribe("MyEvent", event_callback=handle_event_with_delay)

    with ThreadPoolExecutor() as executor:
        event_emitter = ExecutorEventEmitter(executor)
        event_emitter.emit("MyEvent")
        event_emitter.emit("MyEvent")

    print("Closing...")
```

--------------------------------

### FastAPI: Manual Event Emitter Configuration (Python)

Source: https://github.com/mdapena/pyventus/blob/master/docs/learn/events/emitters/fastapi.md

Demonstrates manual setup of Pyventus EventEmitter with FastAPIProcessingService. This approach explicitly passes the background task processor to the EventEmitter, allowing events to be handled asynchronously by FastAPI.

```Python
from fastapi import BackgroundTasks, FastAPI
from pyventus.core.processing.fastapi import FastAPIProcessingService
from pyventus.events import EventEmitter

app = FastAPI()


@app.get("/")
def read_root(background_tasks: BackgroundTasks):
	event_emitter = EventEmitter(event_processor=FastAPIProcessingService(background_tasks))
	event_emitter.emit("MyEvent")
	return {"Event": "Emitted"}
```

--------------------------------

### FastAPI: Factory Method for Event Emitter (Python)

Source: https://github.com/mdapena/pyventus/blob/master/docs/learn/events/emitters/fastapi.md

Shows how to use the FastAPIEventEmitter factory method for a streamlined setup. This method simplifies the integration by automatically configuring the EventEmitter with FastAPI's background tasks.

```Python
from fastapi import Depends, FastAPI
from pyventus.events import EventEmitter, FastAPIEventEmitter

app = FastAPI()


@app.get("/")
def read_root(event_emitter: EventEmitter = Depends(FastAPIEventEmitter())):
	event_emitter.emit("MyEvent")
	return {"Event": "Emitted"}
```

--------------------------------

### Define Sync and Async Event Callbacks

Source: https://github.com/mdapena/pyventus/blob/master/README.md

Pyventus allows defining event callbacks as either synchronous or asynchronous functions. This enables seamless integration into both sync and async Python applications.

```Python
@EventLinker.on("MyEvent")
def sync_event_callback():
    pass  # Synchronous event handling


@EventLinker.on("MyEvent")
async def async_event_callback():
    pass  # Asynchronous event handling
```

--------------------------------

### Customizable Success and Error Handling

Source: https://github.com/mdapena/pyventus/blob/master/README.md

Customize how data streams are handled upon completion or error by configuring complete and error callbacks during the subscription process. This allows for tailored responses to stream events.

```Python
@as_observable_task
async def interactive_counter():
    stop: int = int(input("Please enter a number to count up to: "))  # Can raise ValueError
    for count in range(1, stop + 1):
        yield count
    raise Completed


obs = interactive_counter()
obs.subscribe(
    next_callback=lambda val: print(f"Received: {val}"),
    error_callback=lambda err: print(f"Error: {err}"),
    complete_callback=lambda: print("All done!"),
)
obs()
```

--------------------------------

### Pyventus: Subscribe to All Events with Global Event Listeners

Source: https://github.com/mdapena/pyventus/blob/master/README.md

Shows how to use global event listeners in Pyventus by subscribing to a wildcard pattern (...). This is useful for implementing cross-cutting concerns like logging or monitoring, as it captures all events emitted within a specific EventLinker context.

```Python
@EventLinker.on(...)
def logging(*args, **kwargs):
    print(f"Logging:\n- Args: {args}\n- Kwargs: {kwargs}")


event_emitter: EventEmitter = AsyncIOEventEmitter()
event_emitter.emit("AnyEvent", name="Pyventus")
```

--------------------------------

### Thread Offloading for Observable Tasks

Source: https://github.com/mdapena/pyventus/blob/master/README.md

Demonstrates running observable tasks in separate threads using `concurrent.futures.ThreadPoolExecutor`. This is useful for multithreaded environments where default AsyncIO processing is not suitable.

```Python
from concurrent.futures import ThreadPoolExecutor
from pyventus.reactive import as_observable_task, Completed

@as_observable_task
def simple_counter(stop: int):
    for count in range(1, stop + 1):
        yield count
    raise Completed

if __name__ == "__main__":

    with ThreadPoolExecutor() as executor:

        obs1 = simple_counter(16)
        obs1.subscribe(print)
        obs1(executor)

        obs2 = simple_counter(16)
        obs2.subscribe(print)
        obs2(executor)
```

--------------------------------

### Thread Offloading with Observable Task Execution Context

Source: https://github.com/mdapena/pyventus/blob/master/README.md

Combines thread offloading with the execution context for observable tasks. Tasks are run in separate threads and automatically executed when their respective `with` blocks are exited.

```Python
from concurrent.futures import ThreadPoolExecutor
from pyventus.reactive import as_observable_task, Completed

@as_observable_task
def simple_counter(stop: int):
    for count in range(1, stop + 1):
        yield count
    raise Completed

if __name__ == "__main__":

    with ThreadPoolExecutor() as executor:

        with simple_counter(16).to_thread(executor) as obs1:
            obs1.subscribe(print)

        with simple_counter(16).to_thread(executor) as obs2:
            obs2.subscribe(print)
```

--------------------------------

### Simplified Observable Task Execution Context

Source: https://github.com/mdapena/pyventus/blob/master/README.md

Illustrates how to use observable tasks within a `with` statement block. This enables an execution context that automatically calls the observable tasks upon exiting the block, simplifying manual invocation.

```Python
@as_observable_task
def simple_counter(stop: int):
    for count in range(1, stop + 1):
        yield count
    raise Completed

with simple_counter(stop=16) as obs:
    obs.subscribe(lambda val: print(f"Subscriber 1 - Received: {val}"))
    obs.subscribe(lambda val: print(f"Subscriber 2 - Received: {val}"))
```

--------------------------------

### Display Error Metrics in Python

Source: https://github.com/mdapena/pyventus/blob/master/README.md

This Python snippet demonstrates how to format and display error request metrics. It uses f-strings for output formatting and ANSI escape codes for color highlighting.

```python
import sys

def main():
    # Assume metrics is a dictionary containing counts
    metrics = {
        'error_count': 5
    }
    print(
        f"    - Error Requests: \033[31m{metrics['error_count']}\033[0m"
    )

main()

```

--------------------------------

### Dynamic Voltage Monitoring (Event-Driven Implementation)

Source: https://github.com/mdapena/pyventus/blob/master/README.md

This Python code demonstrates dynamic voltage monitoring using an event-driven architecture with Pyventus. It defines a `VoltageSensor` class that simulates voltage readings within a given range and emits `LowVoltageEvent` or `HighVoltageEvent` based on thresholds. The system uses `AsyncIOEventEmitter` and `EventLinker` to register and handle these events asynchronously, allowing for decoupled logic for voltage protection and notifications.

```Python
import asyncio
import random

from pyventus.events import AsyncIOEventEmitter, EventEmitter, EventLinker


class VoltageSensor:

    def __init__(self, name: str, low: float, high: float, event_emitter: EventEmitter) -> None:
        # Initialize the VoltageSensor object with the provided parameters
        self._name: str = name
        self._low: float = low
        self._high: float = high
        self._event_emitter: EventEmitter = event_emitter

    async def __call__(self) -> None:
        # Start voltage readings for the sensor
        print(f"Starting voltage readings for: {self._name}")
        print(f"Low: {self._low:.3g}v | High: {self._high:.3g}v\n-----------\n")

        while True:
            # Simulate sensor readings
            voltage: float = random.uniform(0, 5)
            print("\tSensor Reading:", "\033[32m", f"{voltage:.3g}v", "\033[0m")

            # Emit events based on voltage readings
            if voltage < self._low:
                self._event_emitter.emit("LowVoltageEvent", sensor=self._name, voltage=voltage)
            elif voltage > self._high:
                self._event_emitter.emit("HighVoltageEvent", sensor=self._name, voltage=voltage)

            await asyncio.sleep(1)


@EventLinker.on("LowVoltageEvent")
def handle_low_voltage_event(sensor: str, voltage: float):
    print(f"🪫 Starting low-voltage protection for '{sensor}'. ({voltage:.3g}v)\n")
    # Perform action for low voltage...


@EventLinker.on("HighVoltageEvent")
def handle_high_voltage_event(sensor: str, voltage: float):
    print(f"⚡ Starting high-voltage protection for '{sensor}'. ({voltage:.3g}v)\n")
    # Perform action for high voltage...


@EventLinker.on("LowVoltageEvent", "HighVoltageEvent")
async def handle_voltage_event(sensor: str, voltage: float):
    print(f"\033[31m\nSensor '{sensor}' out of range.\033[0m (Voltage: {voltage:.3g})")
    # Perform notification for out of range voltage...


async def main():
    # Initialize the sensor and run the sensor readings
    sensor = VoltageSensor(name="CarSensor", low=0.5, high=3.9, event_emitter=AsyncIOEventEmitter())
    await asyncio.gather(sensor(), )  # Add new sensors inside the 'gather' for multi-device monitoring


asyncio.run(main())

```

--------------------------------

### Run Pyventus Version Benchmarks and Generate Plots

Source: https://github.com/mdapena/pyventus/blob/master/docs/release-notes.md

Orchestrates the benchmarking process by installing different pyventus versions, running benchmarks in parallel, and generating comparison plots. It handles console clearing and saving benchmark results to JSON files.

```Python
def main():
    	def clear_console():
    		"""Clear the console based on the operating system."""
    		if os.name == "nt":  # For Windows
    			os.system("cls")
    		else:  # For macOS and Linux
    			os.system("clear")

    	def uninstall_pyventus() -> None:
    		"""Uninstall the currently installed version of pyventus using pip."""
    		subprocess.check_call([sys.executable, "-m", "pip", "uninstall", "-y", "pyventus"])

    	def install_pyventus(version: str = "") -> None:
    		"""Install pyventus using pip, with an optional version parameter."""
    		version = ("." if version == "." else (f"pyventus=={version}" if version else "pyventus"))
    		subprocess.check_call([sys.executable, "-m", "pip", "install", version])

    	# Uninstall the currently installed version of pyventus to avoid conflicts
    	# and ensure a clean environment for the installation of the specified version (0.6.0).
    	uninstall_pyventus()
    	install_pyventus(version="0.6.0")
    	clear_console()

    	# Run the benchmark with pyventus v0.6.0 in a separate process to avoid conflicts.
    	with ProcessPoolExecutor() as executor:
    		fut = executor.submit(run_event_emitter_benchmarks)
    		pyventus_v060_reports = fut.result()
    		clear_console()

    	# Remove executor reference
    	del executor

    	# Uninstall the current version of pyventus again to prepare for the next version.
    	uninstall_pyventus()
    	install_pyventus(version=".")
    	clear_console()

    	# Force a garbage collection.
    	gc.collect()

    	# Run the benchmark with the current version of pyventus (0.7.0) in a separate process.
    	with ProcessPoolExecutor() as executor:
    		fut = executor.submit(run_event_emitter_benchmarks)
    		pyventus_v070_reports = fut.result()
    		clear_console()

    	print("Almost done. Saving reports...")

    	# Save the reports for pyventus v0.6.0 to a JSON file.
    	with open("dist/pyventus_v060_eeb_reports.json", "w") as f1:
    		f1.write(f"{dumps([asdict(report) for report in pyventus_v060_reports])}\n")

    	# Save the reports for pyventus v0.7.0 to a JSON file.
    	with open("dist/pyventus_v070_eeb_reports.json", "w") as f2:
    		f2.write(f"{dumps([asdict(report) for report in pyventus_v070_reports])}\n")

    	clear_console()

    	# Indicate that charts are being generated.
    	print("Generating charts...")

    	# Generate comparison charts for the two versions of pyventus.
    	for i, (v060_report, v070_report) in enumerate(zip(pyventus_v060_reports, pyventus_v070_reports, strict=False)):
    		plot_event_emitter_benchmark_comparison(
    			title=(
    				f"Impact of Subscription Count on Event Emission Time: Comparison of Pyventus v0.6.0 and v0.7.0\n"
    				f"(Event Subscription: {v060_report.event_subscription_mode.capitalize()}, "
    				f"One-time Subscription: {v060_report.onetime_subscription_mode.capitalize()}, "
    				f"Repeats: {v060_report.num_repeats}, Executions: {v060_report.num_executions})\n"
    			),
    			subscription_sizes=v060_report.subscription_sizes,
    			benchmark_bars={

```

--------------------------------

### Factory Method for Redis Event Emitter

Source: https://github.com/mdapena/pyventus/blob/master/docs/learn/events/emitters/redis.md

Shows how to use the RedisEventEmitter factory method for a simplified setup. This method handles the creation of the RedisProcessingService and EventEmitter in one step, requiring only the Redis connection details and the default queue.

```Python
from pyventus.events import EventEmitter, RedisEventEmitter
from redis import Redis
from rq import Queue

redis_conn = Redis.from_url("redis://default:redispw@localhost:6379")
default_queue: Queue = Queue(name="default", connection=redis_conn)

if __name__ == "__main__":
    event_emitter: EventEmitter = RedisEventEmitter(queue=default_queue)
    event_emitter.emit("MyEvent")
```

--------------------------------

### Serve Documentation: Manual vs Hatch

Source: https://github.com/mdapena/pyventus/blob/master/docs/contributing.md

Serve the project documentation in development mode using MkDocs manually or via Hatch for a streamlined development workflow. Hatch manages the serving process efficiently.

```console
mkdocs serve --dev-addr localhost:8000
```

```console
hatch run docs:serve
```