FastAPI example

The example assumes you've already installed our SDK and you're ready to go.
If you wish to see more general info, visit Set up your server article.

Minimal setup

The same way as in the FastAPI docs, the code below can be copied and run immediately. Just make sure you set FISHJAM_URL and FISHJAM_MANAGEMENT_TOKEN environment variables. The Depends function allows the FishjamClient to be provided to the route function as a dependency, which avoids code duplication and allows easy mocking.

import os
from typing import Annotated
from fastapi import Depends, FastAPI
from fishjam import FishjamClient


app = FastAPI()

def fishjam_client():
    fishjam_url = os.environ["FISHJAM_URL"]
    management_token = os.environ["FISHJAM_MANAGEMENT_TOKEN"]
    return FishjamClient(fishjam_url=fishjam_url, management_token=management_token)


@app.get("/")
async def get_rooms(fishjam_client: Annotated[FishjamClient, Depends(fishjam_client)]):
    rooms = fishjam_client.get_all_rooms()
    return {"rooms": rooms}

Listening to events

Fishjam instance is a stateful server that is emitting messages upon certain events. You can listen for those messages and react as you prefer. There are two options for obtaining these.

Webhooks

To receive and parse Fishjam protobuf messages, create a dependable function that parses the request body using the receive_binary function from the fishjam package. Then, you will be able to use pattern matching to respond to different kinds of events.

from fastapi import Depends, FastAPI, HTTPException, Request
from fishjam import  receive_binary
from fishjam.events import ServerMessagePeerAdded

app = FastAPI()

async def parse_fishjam_message(request: Request):
    binary = await request.body()
    if message := receive_binary(binary):
        return message
    raise HTTPException(status_code=400, detail="Invalid Fishjam message")

@app.post("/fishjam-webhook")
async def fishjam_webhook(fishjam_message = Depends(parse_fishjam_message)):
    match fishjam_message:
        case ServerMessagePeerAdded():
            print(f"Peer added: {fishjam_message.peer_id}")
        case _:
            ...

SDK Notifier

Use the asyncio library to run the SDK notifier. It doesn't interact with the FastAPI library per se, just start the event loop and it will run alongside the server.

import asyncio
from jellyfish import FishjamNotifier

notifier = FishjamNotifier(server_address=fishjam_url, server_api_token=management_token)

@notifier.on_server_notification
def handle_notification(notification):
    match notification:
        case ServerMessagePeerAdded():
            print(f"Peer added: {fishjam_message.peer_id}")
        case _:
            ...

async def run_notifier():
  notifier_task = asyncio.create_task(fishjam_notifier.connect())

  # Wait for the notifier to be ready to receive messages
  await fishjam_notifier.wait_ready()

  await notifier_task

asyncio.run(run_notifier())