# Signals

NEW in v21.3

This is a new API being introduced.

This feature is released in BETA.

That means that the API is subject to change, although it is unlikely that it will. It is being released now so that developers may start making use of the feature, and we can continue to refine it towards its final release (currently scheduled for v21.6). Any breaking change from this API will only be done if necessary.

Eventually, it will provide access to plug into the request/response lifecycle.

Signals provide a way for one part of your application to tell another part that something happened.

@app.signal("user.registration.created")
async def send_registration_email(**context):
    await send_email(context["email"], template="registration")
@app.post("/register")
async def handle_registration(request):
    await do_registration(request)
    await request.app.dispatch(
        "user.registration.created",
        context={"email": request.json.email}
    })

# Adding a signal

The API for adding a signal is very similar to adding a route.

async def my_signal_handler():
    print("something happened")
app.add_signal(my_signal_handler, "something.happened.ohmy")

But, perhaps a slightly more convenient method is to use the built-in decorators.

@app.signal("something.happened.ohmy")
async def my_signal_handler():
    print("something happened")

Signals can also be declared on blueprints

bp = Blueprint("foo")
@bp.signal("something.happened.ohmy")
async def my_signal_handler():
    print("something happened")

# Events

Signals are based off of an event. An event, is simply a string in the following pattern:

namespace.reference.action

TIP

Events must have three parts. If you do not know what to use, try these patterns:

  • my_app.something.happened
  • sanic.notice.hello

# Event parameters

An event can be "dynamic" and declared using the same syntax as path parameters. This allows matching based upon arbitrary values.

@app.signal("foo.bar.<thing>")
async def signal_handler(thing):
    print(f"[signal_handler] {thing=}")
@app.get("/")
async def trigger(request):
    await app.dispatch("foo.bar.baz")
    return response.text("Done.")

Checkout path parameters for more information on allowed type definitions.

WARNING

Only the third part of an event (the "action") may be dynamic:

  • foo.bar.<thing> 🆗
  • foo.<bar>.baz

# Waiting

In addition to executing a signal handler, your application can wait for an event to be triggered.

await app.event("foo.bar.baz")

IMPORTANT: waiting is a blocking function. Therefore, you likely will want this to run in a background task.

async def wait_for_event(app):
    while True:
        print("> waiting")
        await app.event("foo.bar.baz")
        print("> event found\n")
@app.after_server_start
async def after_server_start(app, loop):
    app.add_task(wait_for_event(app))

If your event was defined with a dynamic path, you can use * to catch any action.

@app.signal("foo.bar.<thing>")
...
await app.event("foo.bar.*")

# Dispatching

In the future, Sanic will dispatch some events automatically to assist developers to hook into life cycle events.

Dispatching an event will do two things:

  1. execute any signal handlers defined on the event, and
  2. resolve anything that is "waiting" for the event to complete.
@app.signal("foo.bar.<thing>")
async def foo_bar(thing):
    print(f"{thing=}")
await app.dispatch("foo.bar.baz")
thing=baz

# Context

Sometimes you may find the need to pass extra information into the signal handler. In our first example above, we wanted our email registration process to have the email address for the user.

@app.signal("user.registration.created")
async def send_registration_email(**context):
    print(context)
await app.dispatch(
    "user.registration.created",
    context={"hello": "world"}
)
{'hello': 'world'}

FYI

Signals are dispatched in a background task.

# Blueprints

Dispatching blueprint signals works similar in concept to middleware. Anything that is done from the app level, will trickle down to the blueprints. However, dispatching on a blueprint, will only execute the signals that are defined on that blueprint.

Perhaps an example is easier to explain:

bp = Blueprint("bp")
app_counter = 0
bp_counter = 0
@app.signal("foo.bar.baz")
def app_signal():
    nonlocal app_counter
    app_counter += 1
@bp.signal("foo.bar.baz")
def bp_signal():
    nonlocal bp_counter
    bp_counter += 1

Running app.dispatch("foo.bar.baz") will execute both signals.

await app.dispatch("foo.bar.baz")
assert app_counter == 1
assert bp_counter == 1

Running bp.dispatch("foo.bar.baz") will execute only the blueprint signal.

await bp.dispatch("foo.bar.baz")
assert app_counter == 1
assert bp_counter == 2
MIT Licensed
Copyright © 2018-present Sanic Community Organization

~ Made with ❤️ and ☕️ ~