API/Dashboard

class chancy.plugins.api.Api(*, port: int = 8000, host: str = '127.0.0.1', debug: bool = False, allow_origins: list[str] | None = None, secret_key: str | None = None, authentication_backend: AuthBackend)

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.

static get_identifier() → str

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.

async run(worker: Worker, chancy: Chancy)

Start the web server.

class chancy.plugins.api.AuthBackend

Bases: AuthenticationBackend, ABC

abstractmethod async login(request: Request, username: str, password: str) → bool

abstractmethod async logout(request: Request) → None

class chancy.plugins.api.SimpleAuthBackend(users: dict[str, str])

Bases: AuthBackend

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

Parameters:

users – A dictionary of users and their passwords.

async authenticate(conn: HTTPConnection) → tuple[AuthCredentials, BaseUser] | None

async login(request: Request, username: str, password: str) → bool

async logout(request: Request) → None

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

Bases: ABC

A plugin that provides additional API endpoints.

abstractmethod name() → str

Get the name of the plugin.

routes() → list[RouteT]

Get a list of routes to add to the API.

class chancy.plugins.api.plugin.RouteT

Bases: TypedDict

A type hint for a route.

endpoint: Callable

is_websocket: bool | None

methods: 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 get_configuration(request, *, chancy, worker)

Get the configuration of the Chancy instance.

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

Get a single job by ID.

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

Get a list of all the jobs in the system.

Allows ID-based pagination and basic filtering.

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

Get a list of all the queues.

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

Get a list of all the workers.

async login(request: Request, *, chancy, worker)

Login endpoint.

name()

Get the name of the plugin.

routes()

Get a list of routes to add to the API.