Skip to content

FastAPI Event Emitter

  In Pyventus, you can easily integrate Event Emitters with the FastAPI framework through the FastAPI Processing Service. Simply create an instance of the FastAPI Processing Service and pass it as the event processor when setting up the Event Emitter, or you can use the factory method called FastAPI Event Emitter to handle the setup in a single step. 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
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"}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
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"}

  By utilizing the FastAPI Processing Service, the execution of each event emission will be handled by the FastAPI's Background Tasks system.

Practical Example

To start using the Event Emitter with FastAPI, follow these steps:

  1. Install Dependencies: Before proceeding, ensure that you have installed the optional FastAPI dependency.

  2. Subscribe and Emit: Once you have everything installed and configured, using Pyventus with FastAPI is straightforward. Simply define your subscribers and events, instantiate an Event Emitter configured for FastAPI, and emit your events.

    main.py
     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
    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!"}

    To start the FastAPI server, run the following command:

    fastapi dev main.py

    Open your browser and navigate to http://127.0.0.1:8000/email?email_to=email@email.com. You should see the following JSON response:

    { "message": "Email sent!" }

    Additionally, you will be able to view the output of the functions in the console logs.

    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!