Lifespan¶
future.lifespan.Lifespan is the ASGI lifespan hook required by Future. You pass in lists of Tasks to run at startup, on a schedule while the app is up, and at shutdown.
from future.lifespan import Lifespan
from future.interfaces.ITask import ITask
from future.taskscheduler import Unit
from future.application import Future
from app.tasks.ScrapeTask import ScrapeTask
class BootTask(ITask):
name = "boot"
async def run(self) -> None:
...
lifespan = Lifespan(
startup_tasks=[BootTask()],
shutdown_tasks=[],
cron_tasks=[ScrapeTask()],
)
app = Future(lifespan=lifespan, config=config)
Slots¶
| List | When |
|---|---|
startup_tasks |
Once on ASGI startup, in order |
cron_tasks |
Registered with the interval scheduler after startup |
shutdown_tasks |
Once on ASGI shutdown, after the scheduler stops |
Empty lists are fine:
Lifespan(startup_tasks=[], shutdown_tasks=[], cron_tasks=[])
Lifecycle¶
- Run
startup_tasks(await task.run()). - Start
CronSchedulerand registercron_tasks. - App serves traffic; interval tasks fire in the background.
- On shutdown: stop the scheduler, then run
shutdown_tasks.
How to build an ITask (name, interval, jitter, …): see Tasks.