diff --git a/docs/sanic/middleware.md b/docs/sanic/middleware.md index 228c9d47..eafe3c6f 100644 --- a/docs/sanic/middleware.md +++ b/docs/sanic/middleware.md @@ -100,6 +100,20 @@ async def close_db(app, loop): await app.db.close() ``` +It's also possible to register a listener using the `register_listener` method. +This may be useful if you define your listeners in another module besides +the one you instantiate your app in. + +```python +app = Sanic() + +async def setup_db(app, loop): + app.db = await db_setup() + +app.register_listener(setup_db, 'before_server_start') + +``` + If you want to schedule a background task to run after the loop has started, Sanic provides the `add_task` method to easily do so. diff --git a/sanic/app.py b/sanic/app.py index 1513e467..0ef7f784 100644 --- a/sanic/app.py +++ b/sanic/app.py @@ -117,6 +117,19 @@ class Sanic: return decorator + def register_listener(self, listener, event): + """ + Register the listener for a given event. + + Args: + listener: callable i.e. setup_db(app, loop) + event: when to register listener i.e. 'before_server_start' + + Returns: listener + """ + + return self.listener(event)(listener) + # Decorator def route(self, uri, methods=frozenset({'GET'}), host=None, strict_slashes=None, stream=False, version=None, name=None): diff --git a/tests/test_server_events.py b/tests/test_server_events.py index ab0a1fb1..9c0d99eb 100644 --- a/tests/test_server_events.py +++ b/tests/test_server_events.py @@ -49,6 +49,23 @@ def test_single_listener(listener_name): assert random_name_app.name + listener_name == output.pop() +@pytest.mark.parametrize('listener_name', AVAILABLE_LISTENERS) +def test_register_listener(listener_name): + """ + Test that listeners on their own work with + app.register_listener method + """ + random_name_app = Sanic(''.join( + [choice(ascii_letters) for _ in range(choice(range(5, 10)))])) + output = list() + # Register listener + listener = create_listener(listener_name, output) + random_name_app.register_listener(listener, + event=listener_name) + start_stop_app(random_name_app) + assert random_name_app.name + listener_name == output.pop() + + def test_all_listeners(): random_name_app = Sanic(''.join( [choice(ascii_letters) for _ in range(choice(range(5, 10)))]))