Wrapper around APScheduler for managing scheduled tasks
pip install gadfastcron
import contextlib
from fastapi import FastAPI
from gadfastcron import Cron
from gadfastcron import jobstores, triggers
def func():
print('test')
cron = Cron(
jobstores.Sqlalchemy(dsn).store,
(func, triggers.cron.everyday()),
)
@contextlib.asynccontextmanager
async def lifespan(_: FastAPI):
cron.start()
yield
cron.shutdown()
app = FastAPI(lifespan=lifespan)