API/Dashboard

class chancy.plugins.api.Api(*, port=8000, host='127.0.0.1', debug=False, allow_origins=None, secret_key=None, authentication_backend)

Bases: Plugin

Provides an API and a dashboard for viewing the state of the Chancy cluster.

Running this plugin requires a few additional dependencies, you can install them with:

pip install chancy[web]

To have the API run permanently on each Worker, you can add it to the plugins list when creating the Chancy instance:

from chancy.plugins.api import Api
from chancy.plugins.api.auth import SimpleAuthBackend

async with Chancy(..., plugins=[
    Api(
        authentication_backend=SimpleAuthBackend({"admin": "password"}),
        secret_key="<a strong, random secret key>",
    ),
]) as chancy:
    ...

Note

The API is still mostly undocumented, as its development is driven by the needs of the dashboard and may change significantly before it becomes stable.

Since it’s very common to only want the dashboard temporarily, you can start it with the CLI. This can also be used to connect to a remote Chancy instance:

pip install chancy[cli,web]
chancy --app worker.chancy worker web

This will run the API and dashboard on port 8000 by default (you can change this with the --port and --host flags), and generate a temporary random password for the admin user.

Screenshots

param port:

The port to listen on.

param host:

The host to listen on.

param debug:

Whether to run the server in debug mode.

param allow_origins:

A list of origins that are allowed to access the API.

build_starlette_app(worker, chancy)

Build an ASGI app containing the API and dashboard.

Parameters:

• worker (Worker)

• chancy (Chancy)

Return type:

Starlette

static get_identifier()

Returns a unique identifier for this plugin.

This identifier should be unique across all active plugins. If a custom plugin provides compatible functionality to a built-in plugin, it may use the same identifier as the built-in plugin.

Return type:

str

async run(worker, chancy)

Start the standalone web server.

Parameters:

• worker (Worker)

• chancy (Chancy)

Parameters:

• port (int)

• host (str)

• debug (bool)

• allow_origins (list[str] | None)

• secret_key (str | None)

• authentication_backend (AuthBackend)

class chancy.plugins.api.AuthBackend

Bases: AuthenticationBackend, ABC

abstractmethod async login(request, username, password)

Parameters:

• request (Request)

• username (str)

• password (str)

Return type:

bool

abstractmethod async logout(request)

Parameters:

request (Request)

Return type:

None

class chancy.plugins.api.SimpleAuthBackend(users)

Bases: AuthBackend

A simple authentication backend that uses a dictionary of users and passwords.

Parameters:

users (dict[str, str]) – A dictionary of users and their passwords.

async authenticate(conn)

Parameters:

conn (HTTPConnection)

Return type:

tuple[AuthCredentials, BaseUser] | None

async login(request, username, password)

Parameters:

• request (Request)

• username (str)

• password (str)

Return type:

bool

async logout(request)

Parameters:

request (Request)

Return type:

None

class chancy.plugins.api.plugin.ApiPlugin(api)

Bases: ABC

A plugin that provides additional API endpoints.

abstractmethod name()

Get the name of the plugin.

Return type:

str

routes()

Get a list of routes to add to the API.

Return type:

list[RouteT]

class chancy.plugins.api.plugin.RouteT

Bases: TypedDict

A type hint for a route.

endpoint: Callable

is_websocket: bool | None

methods: list[str] | None

name: str | None

path: str

class chancy.plugins.api.core.CoreApiPlugin(api)

Bases: ApiPlugin

Core API plugin which implements the endpoints for jobs, queues, workers, etc…

async static cancel_job(request, *, chancy, worker)

Parameters:

request (Request)

async static get_configuration(request, *, chancy, worker)

Get the configuration of the Chancy instance.

async static get_workers(request, *, chancy, worker)

Get a list of all the workers.

async static health(request, *, chancy, worker)

Health check endpoint - returns 200 if the worker is healthy. Does not require authentication.

Parameters:

request (Request)

async static job(request, *, chancy, worker)

GET: Get a single job by ID. DELETE: Purge a job by ID.

async static jobs(request, *, chancy, worker)

GET: Get a list of jobs with filters and pagination. POST: Batch job actions (retry, purge, cancel).

POST accepts {"action": "retry", "ids": ["job-uuid", ...]}. Each distinct ID is processed independently, in request order. The response contains results with an id, a status of completed, skipped, or failed, and a message for skipped or failed items. ok is true only if every item completed. Earlier successes are not rolled back when a later item fails.

Missing jobs are skipped. Retry skips running jobs; cancel accepts pending, running, and retrying jobs and skips finished jobs. Purge deletes any existing job without cancelling its execution.

GET returns jobs in descending ID order. Pass the last returned ID as before to fetch the next page of lower IDs with the same filters. Completion and scheduling times do not affect this order. Pages are live results, not a snapshot: jobs may be removed or stop matching filters between requests.

Set pagination=true to receive an object with items, has_more, and next_cursor (an ID to pass as before, or null on the last page). Otherwise the response remains a job array. limit must be positive and is capped at 100.

Parameters:

request (Request)

async static jobs_functions(request, *, chancy, worker)

GET: Get a list of distinct function names from jobs. Uses in-memory caching to avoid expensive DISTINCT queries.

Parameters:

request (Request)

async login(request, *, chancy, worker)

Login endpoint.

Parameters:

request (Request)

async logout(request, *, chancy, worker)

Logout endpoint.

Parameters:

request (Request)

name()

Get the name of the plugin.

async static pause_queue(request, *, chancy, worker)

Parameters:

request (Request)

async static plugins(request, *, chancy, worker)

Get detailed information about all registered plugins.

Parameters:

request (Request)

async static queue(request, *, chancy, worker)

Parameters:

request (Request)

async static queues(request, *, chancy, worker)

GET: Get a list of all queues. POST: Create a queue.

Omitted fields use Queue defaults. Omitted tags match all workers; an explicit empty tag list leaves the queue unassigned, so no worker will process its jobs.

Parameters:

request (Request)

async static resume_queue(request, *, chancy, worker)

Parameters:

request (Request)

async static retry_job(request, *, chancy, worker)

Parameters:

request (Request)

routes()

Get a list of routes to add to the API.

async static system_info(request, *, chancy, worker)

Get system information including version, plugins, and database status.

Parameters:

request (Request)

async ws_events(websocket, *, chancy, worker)

WebSocket stream of core system events.

Parameters:

websocket (WebSocket)

chancy.plugins.api.core.parse_filters(filters_param, field_config)

Parse filter triples and build a rule from them.

Args:

filters_param: JSON string of filter triples [[key, op, value], …] field_config: Dict mapping field keys to (Rule, [allowed_operators])

Returns:

tuple: (rule, error_response) - rule: Combined Rule object or None - error_response: Response object if error, None if success