Integrating with application frameworks

WSGI

To integrate APScheduler with web frameworks using WSGI (Web Server Gateway Interface), you need to use the synchronous scheduler and start it as a side effect of importing the module that contains your application instance:

from apscheduler import Scheduler


def app(environ, start_response):
    """Trivial example of a WSGI application."""
    response_body = b"Hello, World!"
    response_headers = [
        ("Content-Type", "text/plain"),
        ("Content-Length", str(len(response_body))),
    ]
    start_response(200, response_headers)
    return [response_body]


scheduler = Scheduler()
scheduler.start_in_background()

Assuming you saved this as example.py, you can now start the application with uWSGI with:

uwsgi --enable-threads --http :8080 --wsgi-file example.py

The --enable-threads (or -T) option is necessary because uWSGI disables threads by default which then prevents the scheduler from working. See the uWSGI documentation for more details.

Note

The Scheduler.start_in_background() method installs an atexit hook that shuts down the scheduler gracefully when the worker process exits.

ASGI

To integrate APScheduler with web frameworks using ASGI (Asynchronous Server Gateway Interface), you need to use the asynchronous scheduler and tie its lifespan to the lifespan of the application by wrapping it in middleware, as follows:

from apscheduler import AsyncScheduler


async def app(scope, receive, send):
    """Trivial example of an ASGI application."""
    if scope["type"] == "http":
        await receive()
        await send(
            {
                "type": "http.response.start",
                "status": 200,
                "headers": [
                    [b"content-type", b"text/plain"],
                ],
            }
        )
        await send(
            {
                "type": "http.response.body",
                "body": b"Hello, world!",
                "more_body": False,
            }
        )
    elif scope["type"] == "lifespan":
        while True:
            message = await receive()
            if message["type"] == "lifespan.startup":
                await send({"type": "lifespan.startup.complete"})
            elif message["type"] == "lifespan.shutdown":
                await send({"type": "lifespan.shutdown.complete"})
                return


async def scheduler_middleware(scope, receive, send):
    if scope['type'] == 'lifespan':
        async with AsyncScheduler() as scheduler:
            await app(scope, receive, send)
    else:
        await app(scope, receive, send)

Assuming you saved this as example.py, you can then run this with Hypercorn:

hypercorn example:scheduler_middleware

or with Uvicorn:

uvicorn example:scheduler_middleware