API cheatsheet¶

import firm.queue as bq
from firm.queue import current_runtime

# bq.configure(database_url, ...) -> Runtime (process-global). All but database_url are keyword-only;
# pass engine=<your SQLAlchemy engine> instead of a URL to share your app's engine/pool:
bq.configure("postgresql://localhost/app", busy_timeout_ms=5000,
             pool_size=20, max_overflow=40,
             default_queue="default", preserve_finished_jobs=True)
current_runtime()        # -> Runtime


@bq.job(queue="default", priority=0, attempts=1, backoff=3.0, concurrency=None)
def my_job(x):           # the function stays directly callable
    ...


my_job.enqueue(x)                    # -> int | None  (job id, or None if discarded)
my_job.enqueue_at(when, x)           # when: datetime
my_job.enqueue_in(delta, x)          # delta: timedelta

concurrency = {
    "key": lambda *a, **k: a[0],  # the variable part of the concurrency key
    "to": 1,                      # max simultaneous executions per key
    "duration": 300,              # seconds before the semaphore expires (failsafe)
    "group": None,                # share a key namespace across jobs (default: this job)
    "on_conflict": "block",       # "block" (queue it) or "discard" (drop it)
}

from firm.queue import queues

queues.all_queues(rt)    # -> list[str]
queues.size(rt, q)       # -> int
queues.pause(rt, q)
queues.resume(rt, q)
queues.is_paused(rt, q)  # -> bool
queues.clear(rt, q)      # -> int  (jobs removed)
queues.latency(rt, q)    # -> float  (seconds since the oldest ready job)

from datetime import timedelta
from firm.queue import maintenance

maintenance.retry_failed(rt, job_id)                          # -> bool
maintenance.retry_all_failed(rt)                              # -> int
maintenance.clear_finished(rt, older_than=timedelta(days=7), batch_size=500)  # -> int

# CLI (production):
#   firm-queue start --import myapp.jobs [--mode fork|thread] [--threads N] [--queues a,b]

from firm.queue import current_runtime
from firm.queue.supervisor import (ThreadSupervisor, ForkSupervisor, SupervisorConfig,
                                     WorkerConfig, DispatcherConfig)

config = SupervisorConfig(workers=[WorkerConfig(queues=("*",), threads=3)],
                          dispatchers=[DispatcherConfig()], recurring=[])
sup = ThreadSupervisor(current_runtime(), config)   # in-process; .start() is non-blocking
sup.start()
sup.stop()                                          # or:  with ThreadSupervisor(rt, config): ...
ForkSupervisor(current_runtime(), config).start()   # production default (POSIX); blocks

from firm.queue.worker import run_ready
from firm.queue.dispatcher import dispatch_once, run_maintenance

run_ready(rt, queues=("*",), limit=10)    # -> int  (claim + run one batch of up to `limit`)
dispatch_once(rt)                          # -> int  (promote due scheduled jobs)
run_maintenance(rt)                         # -> int  (release blocked jobs with capacity)

from firm.queue.scheduler import RecurringTask, Scheduler

# 5-field cron; pass a list as SupervisorConfig(recurring=[...]) to start a scheduler automatically
task = RecurringTask(key="nightly", schedule="0 3 * * *", job=my_job, args=(), kwargs={})
Scheduler(rt, [task]).tick()   # manual: enqueue anything due this period

from firm.queue import queries

queries.STATES                                     # the valid `state` values, in display order
queries.state_counts(conn, queue=None)             # -> dict[str, int]  (per state, plus total)
queries.queue_rows(conn, now)                      # -> list[dict]  (name, size, latency, paused)
queries.queue_names(conn)                          # -> list[str]  (queues with ready jobs)
queries.queue_size(conn, queue)                    # -> int
queries.queue_latency(conn, queue, now)            # -> float  (seconds, 0.0 if empty)
queries.jobs_by_state(conn, state, limit=50, offset=0, queue=None)   # -> list[dict]
queries.job_detail(conn, job_id)                   # -> dict | None
queries.processes(conn, now, alive_threshold=300.0)  # -> list[dict]  (+ age, alive)
queries.recurring(conn)                            # -> list[dict]
# a negative limit/offset, or an unknown state, raises ValueError.

from firm.cache import Cache, JSONCoder, MsgpackCoder, PickleCoder

# database_url is positional; everything after it is keyword-only (defaults shown):
cache = Cache("sqlite:///cache.db", engine=None, coder=None, encrypt_key=None,
              max_age=1209600.0, max_size=268435456, max_entries=None,   # 2 weeks / 256 MiB / off
              expiry_batch_size=100, max_key_bytesize=1024, size_estimate_samples=10000,
              create_schema=True, auto_expire=True, background_expiry=False, expiry_interval=60.0)

cache.get(key)                      # -> Any | None
cache.set(key, value)
cache.fetch(key, default_or_callable)   # -> Any  (compute + store on a miss)
cache.delete(key)                   # -> bool
cache.exist(key)                    # -> bool
cache.get_multi(keys)               # -> dict
cache.set_multi(mapping)
cache.increment(key, by=1)          # -> int
cache.decrement(key, by=1)          # -> int
cache.clear()
cache.close()                       # or:  with Cache(...) as cache: ...
# keys are str|bytes; values go through the coder (JSONCoder by default; MsgpackCoder for
# compact binary rows, PickleCoder for arbitrary Python objects — see cache/encryption-and-coders).
# encrypt_key=<Fernet key> encrypts values at rest (needs the [encryption] extra).

from firm.cache import queries

queries.cache_stats(conn)                        # -> {"entries": int, "estimated_size": int}
queries.cache_recent(conn, limit=50, offset=0)   # -> list[dict]  (id, key, byte_size, created_at)
# `key` is raw bytes (a cache key is arbitrary bytes — decode it yourself for display);
# a negative limit/offset raises ValueError.

from firm.channel import Channel

channel = Channel("sqlite:///cable.db", engine=None, polling_interval=0.1,
                  message_retention=86400.0, auto_trim=True, trim_batch_size=100,
                  create_schema=True)

channel.broadcast(name, payload)    # name/payload: str|bytes (str payloads are UTF-8 encoded)
channel.subscribe(name, callback)   # callback(payload: bytes); only future messages; per-process
channel.unsubscribe(name, callback)
channel.trim()                      # -> int
channel.close()                     # or:  with Channel(...) as channel: ...

from firm.channel import queries

queries.channel_stats(conn)                       # -> {"messages": int, "channels": int}
queries.channel_top(conn, limit=25, offset=0)     # -> list[dict]  (channel, count, last)
queries.channel_recent(conn, limit=50, offset=0)  # -> list[dict]  (id, channel, payload, created_at)
# `channel` and `payload` are raw bytes (decode them yourself for display);
# a negative limit/offset raises ValueError.

from firm.audit import AuditLog, Ref, record

# database_url is positional; everything after it is keyword-only (defaults shown):
audit = AuditLog("sqlite:///audit.db", engine=None, create_schema=True,
                 max_age=None, background_retention=False, retention_interval=3600.0)

audit.record(action, subject=None, actor=None, data=None, changes=None,
             correlation_id=None, context=None, conn=None)
audit.history(subject=None, subject_type=None, subject_id=None,
              actor=None, actor_type=None, actor_id=None, action=None,
              correlation_id=None, since=None, limit=100)   # -> list[dict]
# subject_type=/subject_id= filter independently (either alone, or both); same for actor_*.
# pass either subject= or subject_type=/subject_id= for a field, never both (raises ValueError).
# a bare-string filter (actor="cron") filters by type only.
audit.close()                       # or:  with AuditLog(...) as audit: ...

# shared-DB, same-transaction (atomic with the business change):
record(conn, action, subject=None, actor=None, data=None, changes=None,
       correlation_id=None, context=None)

# subject/actor accept the same forms — type and id are each optional:
#   invoice                     domain object with `.id`  -> (ClassName, id)
#   ("Invoice", 42)             tuple; either half may be None (empty/None id -> NULL)
#   "cron"                      bare string -> a role/kind label, stored as the type (no id)
#   Ref("User", 7, name="…")    explicit, with an optional human-readable display name
#   None                        no actor/subject (a system event)
# An object may define __firm_audit_ref__(self) -> Ref to customize its audit identity.
# Each event dict from history()/get() carries subject_label/actor_label (the display names).
# data/changes/context: dicts stored as JSON text (not native JSON/JSONB).

from firm.audit import queries

queries.audit_stats(conn)            # -> {"events": int, "actions": int, "last_event_at": dt|None}
queries.audit_count(conn, action=None, subject=None, subject_type=None, subject_id=None,
                    actor=None, actor_type=None, actor_id=None, correlation_id=None)   # -> int
queries.audit_search(conn, **same_filters, sort="created_at", dir="desc",
                     limit=25, offset=0)   # -> list[dict]  (the history() shape + row_mac)
queries.audit_detail(conn, event_id)       # -> dict | None
queries.SORT_COLUMNS, queries.DEFAULT_SORT # the sortable columns and the default
# filters behave exactly as in history() (a bare string filters by type only, and passing both
# forms for one field raises ValueError); an unknown sort/dir falls back to the default, and a
# negative limit/offset raises ValueError.

# tamper-evidence (only meaningful when a key is configured; see audit/tamper-evidence):
queries.verify_status_row(conn)                       # -> dict | None  (the verifier's status row)
queries.integrity_config(conn, key_configured=True)   # -> IntegrityConfig
queries.integrity_state(status, config, now=None,
                        verify_max_age=..., anchor_max_age=...)   # -> IntegrityState (pure)
queries.row_integrity_context(conn)   # -> dict  (the per-row signals, gathered once)
queries.row_status(row, ctx)          # -> "sealed"|"unsealed"|"unprotected"|"tampered"|... | None
# IntegrityState carries state, tone (ok/warn/danger/neutral), escalate, and cause tokens.

# FastAPI: app = FastAPI(lifespan=lifespan(database_url=..., embed_workers=False,
#                                          queues=("*",), threads=3))
from firm.queue.contrib.fastapi import lifespan

# Flask: FirmQueue(app, database_url=None, embed_workers=False, queues=("*",), threads=3)
#   reads app.config/env FIRM_QUEUE_DATABASE_URL, then FIRM_DATABASE_URL;
#   registers `flask firm-queue worker`
from firm.queue.contrib.flask import FirmQueue

from firm.queue.contrib.sqlalchemy import enqueue_after_commit
enqueue_after_commit(session, my_job, x)   # enqueues iff the session commits

from firm.ui import BasicAuth, build_dashboard, serve, static_dir
dash = build_dashboard(database_url=None, queue_url=None, cache_url=None,
                       channel_url=None, audit_url=None)   # -> Dashboard (close() when done)
serve(dash, host="127.0.0.1", port=8787, authenticator=None, channel_trim_retention=None)

# the dashboard without a transport: handle(UIRequest) -> UIResponse
from firm.ui import DashboardApp, Headers, UIRequest, UIResponse
app = DashboardApp(dash, authenticator=None, channel_trim_retention=None, static_url=None)
app.handle(UIRequest("GET", "/jobs", query="state=failed", headers=Headers(...),
                     body=b"", peer="", prefix="/firm", host="app.example"))
static_dir()   # -> Path  (the stylesheet's directory, to serve it from your own pipeline)

from firm.ui.contrib.django import dashboard_urls, dashboard_view   # decorator=…, host_auth=…
from firm.ui.contrib.flask import blueprint                          # -> Blueprint
from firm.ui.contrib.fastapi import router                           # -> APIRouter

firm-queue start|work|drain|dispatch|maintenance --database-url ... --import myapp.jobs
firm-cache stats|clear|trim --database-url ...
firm-channel stats|trim --database-url ...
firm-audit stats|history|prune --database-url ...
firm-ui --database-url ... [--queue-url ... --cache-url ... --channel-url ... --audit-url ...] [--host --port]