Use with Fastapi

While you can keep your Chancy worker separate from your FastAPI application, you can also run your Chancy worker in the same process as your FastAPI application by using FastAPI’s lifespan events:

import asyncio
import contextlib
from typing import AsyncIterator

from fastapi import FastAPI
from chancy import Chancy, Worker, Queue, job

chancy = Chancy("postgresql://localhost/postgres")

@contextlib.asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """
    FastAPI lifespan handler that starts and stops the Chancy worker.
    This ensures the worker starts when FastAPI starts and shuts down properly.
    """
    # Run the database migrations (don't do this in production)
    await chancy.migrate()

    # Declare any queues we need.
    await chancy.declare(Queue("default"))

    # Start the worker in the background and return control to FastAPI
    async with Worker(chancy) as worker:
      yield


app = FastAPI(lifespan=lifespan)


@job(queue="default")
async def send_an_email():
    print("Sending an email")


@app.get("/")
async def read_root():
    await chancy.push(send_an_email)
    return {"Hello": "World"}

This can be useful for small applications and simple deployments (like containers meant for UnRAID) where you don’t want to manage multiple processes or containers. However, for larger applications, it’s recommended to keep your worker separate from your FastAPI application.

Mount the dashboard

You can also serve the Chancy dashboard and API through your existing FastAPI application. Install chancy[web] alongside FastAPI, then use build_starlette_app() to build the sub-application:

from contextlib import asynccontextmanager

from fastapi import FastAPI
from chancy import Chancy, Worker
from chancy.plugins.api import Api, SimpleAuthBackend

chancy = Chancy("postgresql://localhost/postgres")
dashboard = Api(
    authentication_backend=SimpleAuthBackend({"admin": "change-me"}),
    secret_key="replace-with-a-strong-random-secret",
)
worker = Worker(chancy, register_signal_handlers=False)

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with chancy:
        async with worker:
            yield

app = FastAPI(lifespan=lifespan)
app.mount("/chancy", dashboard.build_starlette_app(worker, chancy))

Let the parent ASGI server handle process signals by passing register_signal_handlers=False to the embedded worker. Its lifespan context will stop the worker when the server shuts down.