celery-farm 0.1.1

Generate OpenAPI/Swagger docs and REST endpoints for your registered Celery tasks & beat schedule.
Integrates with FastAPI, Django, and Flask.

The preview above is an illustration. To capture real screenshots, run an example app (see Run the examples), open /celery/docs and /beat/docs, and drop the PNGs into docs/assets/ (replacing these references).

Installation

# pip
pip install celery-farm              # core only
pip install "celery-farm[fastapi]"   # or [django] / [flask]

# uv
uv add celery-farm
uv add "celery-farm[fastapi]"        # or [django] / [flask]

The framework integration is an optional extra — install the one you use (fastapi, django, or flask). The core (build_openapi, task introspection) needs only celery + pydantic.

Overview

Every integration exposes the same routes:

• POST /tasks/{name} — queue a task, returns its task_id.
• GET /tasks — list registered tasks, parameters, and return types.
• GET /results/{task_id} — poll task status/result.
• GET /openapi.json + GET /docs — OpenAPI document and Swagger UI (tasks).

The periodic-task beat schedule lives in its own separate Swagger (/beat/docs), so the main docs stay focused on task calls — see Beat schedule.

Per task you can override OpenAPI fields via the decorator: @app.task(summary=..., description=..., tags=[...], deprecated=True, openapi_extra={...}). A task with a single object-typed argument (pydantic model / TypedDict / dataclass / dict) takes that object directly as the request body.

Tasks typed with PEP 484 type comments (instead of annotations) are also supported — celery_farm falls back to parsing # type: comments, so the params are still typed in the schema and validated. See demo.legacy_mul in examples/fastapi_app/app.py:

@celery_app.task(name="demo.legacy_mul", tags=["legacy"])
def legacy_mul(x, y):
    # type: (int, int) -> int
    return x * y

Beat schedule (separate Swagger)

The beat schedule is documented in its own Swagger UI, separate from the task-calling docs, with two kinds of endpoint:

• GET /schedule — read-only index listing every entry (task, cadence, args).
• POST /schedule/{entry} — runs the entry now: dispatches its task immediately with the configured args/kwargs and returns a task_id (same shape as POST /tasks/{name}). Each entry is its own operation, so it shows up in Swagger with the task's summary/description.

(The recurring execution itself is still driven by the celery beat process — these endpoints let you inspect the schedule and trigger an entry on demand.) Guard the POST with dependencies= (FastAPI) or your framework's auth as needed.

An entry's summary/description default to the scheduled task's own summary/description, so create_beat_app(celery_app) documents the schedule with no extra config. beat_meta is optional — use it to override summary, description, tags, deprecated, and openapi_extra per entry (mirroring the @app.task(...) decorator); a Celery beat_schedule entry can't carry these keys itself, so they're passed here:

create_beat_app(celery_app, beat_meta={
    "ping-every-30s": {
        "summary": "Liveness ping",
        "description": "Every 30s…",
        "tags": ["health"],        # groups the entry under a "health" section
        "deprecated": True,        # also inherited from the task if it's deprecated
        "openapi_extra": {"externalDocs": {"url": "https://…"}},  # any operation field
    },
})  # get_urlpatterns(...) / create_blueprint(...) accept beat_meta too

summary/description/deprecated fall back to the referenced task's own values when not given, so a @app.task(deprecated=True) marks its scheduled entry deprecated automatically.

• FastAPI — mount tasks and beat as separate sub-apps:

  from celery_farm import create_beat_app, create_task_app
  
  app.mount("/celery", create_task_app(celery_app))
  app.mount("/beat", create_beat_app(celery_app))
  # tasks:  /celery/docs, POST /celery/tasks/{name}, ...
  # beat:   /beat/docs, GET /beat/schedule, POST /beat/schedule/{entry}, /beat/openapi.json
  

• Django / Flask — automatic: get_urlpatterns() / create_blueprint() already add /…/beat/docs (+ GET /…/beat/schedule, POST /…/beat/schedule/{entry}, and /…/beat/openapi.json) alongside the main /…/docs.

pydantic v1 / v2

Both are supported (pydantic>=1.9). celery_farm detects the installed version and adapts: it emits OpenAPI 3.1 under pydantic v2 and 3.0 under v1 (matching each version's JSON Schema dialect).

Two version-specific caveats under pydantic v1:

• Celery's @task(pydantic=True) model reconstruction requires v2. Under v1 a model-annotated argument arrives as a plain dict.
• The FastAPI integration's v1 support is bounded by FastAPI itself — recent FastAPI is v2-only, so pair pydantic v1 with a FastAPI release that still supports it (e.g. fastapi==0.111). The Django/Flask integrations have no such constraint.

One caveat under pydantic v2 on Python < 3.12: a task whose argument is a TypedDict must use typing_extensions.TypedDict (not typing.TypedDict) — pydantic v2 requires this to build the schema. On Python 3.12+ either works.

All combinations (Python 3.11–3.14 × pydantic v1/v2) are exercised in CI.

Quickstart (FastAPI)

from celery import Celery
from fastapi import FastAPI
from celery_farm import create_beat_app, create_task_app

celery_app = Celery("demo")

@celery_app.task(name="demo.add")
def add(x: int, y: int) -> int:
    return x + y

app = FastAPI()
# Tasks and beat are each mounted as their own sub-app with isolated Swagger UI.
app.mount("/celery", create_task_app(celery_app))  # /celery/docs, /celery/tasks/{name}
app.mount("/beat", create_beat_app(celery_app))    # /beat/docs, /beat/schedule

Both sub-apps get their own native FastAPI Swagger (/celery/docs, /beat/docs). dependencies= applies FastAPI Depends (e.g. auth) to every route:

app.mount("/celery", create_task_app(celery_app, dependencies=[Depends(auth)]))

Any other keyword is forwarded to the underlying FastAPI(...) (typed via PEP 692 Unpack, so common options autocomplete). Disable the docs/schema, or set any app option:

app.mount("/celery", create_task_app(celery_app, docs_url=None, openapi_url=None))
app.mount("/beat", create_beat_app(celery_app, version="2.0.0"))

FastAPI renders Swagger UI natively, so customize the viewer with FastAPI's own swagger_ui_parameters= (the equivalent of Django/Flask's swagger_ui_options):

app.mount("/celery", create_task_app(
    celery_app,
    swagger_ui_parameters={"docExpansion": "none", "filter": True},
))

To point Swagger assets at a different CDN (the FastAPI counterpart of swagger_cdn), FastAPI has no constructor option — override the /docs route with fastapi.openapi.docs.get_swagger_ui_html(..., swagger_js_url=..., swagger_css_url=...).

Making sure every task is exposed (finalize=)

Routes are built from a one-time snapshot of celery_app.tasks, so only tasks already registered at that moment are exposed. A Celery worker imports all task modules (from the imports/include config or autodiscover_tasks) at startup, but a web process does not — so tasks whose modules haven't been imported yet would be silently missing from the docs and routes.

To avoid that, every entry point (create_task_app, create_beat_app, build_openapi, get_urlpatterns, create_blueprint) runs that same discovery by default (finalize=True) before building the app — so it "just works":

app.mount("/celery", create_task_app(celery_app))  # imports task modules first
app.mount("/beat", create_beat_app(celery_app))

Because this imports your task modules, an import error surfaces when the app is built (generally what you want — a task that can't be imported can't run in a worker either). Pass finalize=False to skip discovery and expose only what's already registered, e.g. when you import the task modules yourself or want to avoid the import side effects. The equivalent manual call is celery_farm.finalize_app(celery_app) (or Celery's celery_app.loader.import_default_modules()).

Quickstart (Django)

# urls.py
from django.urls import include, path
from celery_farm.integrations.django import get_urlpatterns
from myproj.celery import app as celery_app

urlpatterns = [
    path("celery/", include(get_urlpatterns(celery_app))),
]
# tasks: /celery/docs  (POST /celery/tasks/{name}, /celery/tasks, /celery/results/{id})
# beat:  /celery/beat/docs  (GET /celery/beat/schedule, POST /celery/beat/schedule/{entry})

Pure Django (no DRF). Request bodies are validated with pydantic; the OpenAPI document is built by celery_farm.build_openapi and Swagger UI is loaded from a CDN. build_openapi(celery_app) is also usable standalone to just extract a spec.

Quickstart (Flask)

from flask import Flask
from celery_farm.integrations.flask import create_blueprint
from myproj.celery import app as celery_app

app = Flask(__name__)
app.register_blueprint(create_blueprint(celery_app), url_prefix="/celery")
# tasks: /celery/docs  (POST /celery/tasks/{name}, /celery/tasks, /celery/results/{id})
# beat:  /celery/beat/docs  (GET /celery/beat/schedule, POST /celery/beat/schedule/{entry})

Customizing Swagger UI (Django / Flask)

The Django/Flask integrations render Swagger UI from a CDN. Two knobs let you change where the assets load from and how the viewer behaves — both apply to the task and beat docs:

• swagger_cdn — base URL for swagger-ui.css / swagger-ui-bundle.js (pin a version, swap to unpkg, or self-host). Defaults to https://cdn.jsdelivr.net/npm/swagger-ui-dist@5.
• swagger_ui_options — a dict merged into the SwaggerUIBundle({...}) call (e.g. docExpansion, filter, tryItOutEnabled, defaultModelsExpandDepth). Values must be JSON-serializable, so JS-function options are not supported.

# Flask
create_blueprint(
    celery_app,
    swagger_cdn="https://unpkg.com/swagger-ui-dist@5.11.0",
    swagger_ui_options={"docExpansion": "none", "filter": True},
)

# Django
get_urlpatterns(
    celery_app,
    swagger_cdn="https://unpkg.com/swagger-ui-dist@5.11.0",
    swagger_ui_options={"docExpansion": "none", "filter": True},
)

The same options are available on the low-level celery_farm.openapi.swagger_ui_html helper (cdn= / swagger_ui_options=) if you render the page yourself.

FastAPI renders Swagger natively — customize it there with FastAPI's own swagger_ui_parameters= (and other options) forwarded through create_task_app / create_beat_app.

Run the examples

# FastAPI
uv sync --extra fastapi
uv run uvicorn examples.fastapi_app.app:app --reload   # /docs  +  /beat/docs

# Django
uv sync --extra django
PYTHONPATH=. DJANGO_SETTINGS_MODULE=examples.django_app.settings \
    uv run django-admin runserver                      # http://localhost:8000/celery/docs

# Flask
uv sync --extra flask
uv run flask --app examples.flask_app.app run          # http://localhost:5000/celery/docs

Development

make test        # tests on the current interpreter (pydantic v2)
make test-v1     # current interpreter, pydantic v1 (pinned FastAPI/httpx)
make test-all    # full matrix: Python 3.11–3.14 × pydantic v1/v2 (uv fetches each)

CI (.github/workflows/ci.yml) runs the same 3.11–3.14 × v1/v2 matrix plus a pre-commit lint job on every push and PR.

License

Released under the MIT License — © 2026 milksys.