Stirfried 🥡

Stirfried is an ASGI HTTP/Socket.IO server that provides both browser-based and regular clients with real-time control over Celery tasks.

Tasks are scheduled by name, meaning the server won't necessarily need an update when changes are made to the workers and tasks available to it.

Stirfried implements a simple-to-scale, three layered architecture: clients, servers and workers. Any layer can be scaled out by adding more instances.

Stirfried provides Socket.IO and HTTP APIs with three core functions:

• Schedule a task
• Revoke a task
• Query task info

Want to see Stirfried in action before digging through the README? Try running the example.

Built on:

• starlette & uvicorn
• python-socketio
• celery

Workers

Install Stirfried in your Celery workers via pip/pipenv/poetry:

pip install stirfried

Import the StirfriedTask:

from stirfried.celery import StirfriedTask

Configure the base class globally:

app = Celery(..., task_cls=StirfriedTask)

...or per task:

@app.task(base=StirfriedTask)
def add(x, y, room=None):
    return x + y

Servers

The server can be run by running the korijn/stirfried Docker container and exposing port 8000, or alternatively by cloning this repo, installing the dependencies with poetry and starting the uvicorn server as demonstrated in the example code.

You can configure both the servers and workers with a settings.py file, via the standard Celery configuration mechanism. In the docker deployment scenario, you can mount the settings file to the path /app/settings.py.

Clients

Clients can connect to the Socket.IO API using standard Socket.IO libraries, and to the HTTP API using plain window.fetch.

Task object schema

Tasks are scheduled by submitting the following task object to either of the APIs:

{
    "task_name": "",  // (required) task name
    "args": [],       // (optional) task arguments
    "kwargs": {},     // (optional) task keyword arguments
    "room": "",       // (optional) custom room override, only processed if
                      //            `custom_rooms` is enabled
                      //            NOTE: can also be used to disable server events
                      //                  for this task by passing the sentinel room
                      //                  ("NO_EMIT" by default)
    "chain": []       // (optional) array of task objects to chain onto the main task
                      //            task objects use the same schema, except for
                      //            the `chain` property which cannot be nested further
                      //            NOTE: chained tasks are applied in reverse order
}

Socket.IO API

Events are described in the following format: name(args[, optional]) -> callback_args

Clients can emit any of the following events that servers are listening for:

Clients can listen for the following server-emitted events, which directly hook into the Celery Task class callbacks, except for on_progress which is a Stirfried addition and may be implemented by tasks to support progress events:

HTTP API

Settings

You can configure Celery, Socket.IO and Stirfried all from the same settings.py file. Stirfried settings are prefixed with stirfried_, Socket.IO settings are prefixed with socketio_, and Celery settings are used as-is (not prefixed).

Socket.IO server settings are passed on directly (but without the prefix) to the AsyncServer constructor of the python-socketio library, see their documentation for the options that are available. See the Celery documentation for the options there.

The following options are additionally available for configuring Stirfried servers and workers:

Rooms

For Socket.IO API scheduled tasks, server-emitted events are sent to the client that scheduled the task by default. For HTTP API scheduled tasks, server-emitted events are not emitted by default. The server accomplishes this by injecting a value into the room keyword argument of Stirfried Celery tasks.

The StirfriedTask base class depends on the presence of this keyword argument.

This means you are required to add the keyword argument room=None to your task definitions in order to receive it.

If custom_rooms is enabled, clients can override the value by sending along a custom room value (though not via the keyword arguments of the task, see the API schema documentation).

Progress events

You can emit progress events from workers by calling self.emit_progress(current, total, info=None) in a task.

You can use the optional info keyword argument to send along arbitrary metadata, such as a progress message or early results.

Note that you are required to pass bind=True to the celery.task decorator in order to get access to the self instance variable.

@celery.task(bind=True)
def add(self, x, y, room=None):
    s = x
    self.emit_progress(50, 100)  # 50%
    s += y
    return s

Binary/big data

Socket.IO clients, servers and Celery workers support the msgpack transport, allowing you to use binary data directly (without needing to manually convert to and from base64 encoded strings and suffering the according performance penalty).

You should also be aware of limitations in Redis on client output buffers. This means that you cannot emit events greater than a certain value (32mb by default). You can override this setting in various ways, here's how to do it via the Redis server CLI:

# client-output-buffer-limit <class> <hard limit> <soft limit> <soft seconds>
redis-server --client-output-buffer-limit pubsub 256mb 128mb 30

Testing

When unit testing a Stirfried Celery worker, the recommended approach is to disable the Redis connection by simply leaving out stirfried_redis_url from your settings, and to directly call the task functions in unit tests. The lack of a Redis connection will short-circuit any events that would normally be emitted. This setup will allow you to treat tasks as regular functions and perform unit testing as usual.

Optionally, you can patch/mock any calls to self.emit_progress using standard Python testing utilities to test those too.

Example code

The repo includes an example demonstrating all of the functionality provided by Stirfried.

You can run the example as follows:

• Clone the repository
• cd into the example directory
• Run docker-compose build
• Then docker-compose up
• Open your browser and go to http://localhost:8080/
• You should see the following interface and are ready to give Stirfried a try: