Framework integration¶

firm.queue.contrib has optional glue for embedding firm-queue in a web app. Each piece is opt-in, behind its own extra, and nothing in core imports it — if you don't use it, it costs nothing. You still define jobs the normal way with @bq.job.

There is no firm.cache.contrib.flask or firm.channel.contrib.fastapi, and that is deliberate. The queue needs framework glue because it has process-global state and background threads that must be tied to the app lifecycle; Cache, Channel and audit.record are plain objects you construct yourself, so there is nothing to hook. Django is the exception — it has pluggable backend contracts to fill — which is why the integrations for the other modules start there.

The dashboard has framework glue of its own — firm.ui.contrib mounts it in a Django, Flask, or FastAPI application instead of running it as a second process. See Dashboard § Mount it in your application.

FastAPI¶

from fastapi import FastAPI
from firm.queue.contrib.fastapi import lifespan

app = FastAPI(lifespan=lifespan(database_url="postgresql://localhost/app"))

@app.post("/welcome/{user_id}")
def welcome(user_id: int):
    send_welcome.enqueue(user_id)      # a normal @bq.job
    return {"queued": True}

The lifespan calls configure(...) on startup so your handlers can enqueue. Pass embed_workers=True (with queues=, threads=) to also run a worker + dispatcher inside the app process — convenient for development or a single-process deploy; it's stopped on shutdown.

Flask¶

from flask import Flask
from firm.queue.contrib.flask import FirmQueue

app = Flask(__name__)
app.config["FIRM_QUEUE_DATABASE_URL"] = "postgresql://localhost/app"
FirmQueue(app)                    # or FirmQueue(app, database_url="...")

@app.post("/welcome/<int:user_id>")
def welcome(user_id):
    send_welcome.enqueue(user_id)
    return "", 202

The URL is looked up in database_url=, then app.config, then the environment; within each of those the queue's own FIRM_QUEUE_DATABASE_URL wins over the suite-wide FIRM_DATABASE_URL, so one shared setting still configures every module while a single module can override it.

The extension configures the queue and registers a CLI group, so you run workers with:

flask firm-queue worker --threads 5 --queues default,mailers

FirmQueue(app, embed_workers=True) runs the worker inside the web process instead (dev / single-process only — otherwise every web worker starts its own supervisor).

Transactional enqueue¶

enqueue_after_commit defers an enqueue until your SQLAlchemy session commits, and drops it on rollback — so you never enqueue a job for a request that didn't persist:

from firm.queue.contrib.sqlalchemy import enqueue_after_commit

def create_order(session, payload):
    order = Order(**payload)
    session.add(order)
    enqueue_after_commit(session, charge_card, order.id)   # fires iff the commit succeeds
    session.commit()

The job is enqueued in firm's own transaction just after your commit — so it's "enqueue iff the request committed," not one atomic transaction. A crash in the narrow window between the two commits could still drop the enqueue; for most apps that's the right trade.

Production shape¶

Embed workers for dev; in production run them as separate processes (firm-queue start or flask firm-queue worker) so you scale web and worker capacity independently. See Workers & the supervisor.