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