How to ensure codes in FastAPI's lifespan is run in only one of the workers for Uvicorn?

Question

I would like to add a scheduled job in FastAPI and I have put the scheduler setup and shutdown codes in the lifespan of the FastAPI app. However, I notice that both workers are running the scheduler independently.

Is there any way I can ensure only one worker works on the scheduled job?

Sample code:

from fastapi import FastAPI
from datetime import datetime
from contextlib import asynccontextmanager
from apscheduler.schedulers.asyncio import AsyncIOScheduler
import uvicorn

def test():
    print(f"Test scheduler {datetime.now()}")
    
@asynccontextmanager
async def lifespan(app: FastAPI):
    scheduler = AsyncIOScheduler() 
    
    scheduler.add_job(test, trigger="cron", second="0-30")
    scheduler.start()
        
    yield
    
    scheduler.shutdown()
    
app = FastAPI(lifespan=lifespan)

if __name__ == "__main__":
    uvicorn.run("main:app", workers=2)

Output, where test() is run twice per second when I only want it to run once per second:

INFO:     Uvicorn running on http://127.0.0.1:5000 (Press CTRL+C to quit)
INFO:     Started parent process [34952]
INFO:     Started server process [34957]
INFO:     Waiting for application startup.
INFO:     Started server process [34958]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Application startup complete.
Test scheduler 2024-02-21 14:51:15.001191
Test scheduler 2024-02-21 14:51:15.001314
Test scheduler 2024-02-21 14:51:16.000480
Test scheduler 2024-02-21 14:51:16.001643
Test scheduler 2024-02-21 14:51:17.000901
Test scheduler 2024-02-21 14:51:17.002765
INFO:     Shutting down
INFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     Waiting for application shutdown.
INFO:     Application shutdown complete.
INFO:     Application shutdown complete.
INFO:     Finished server process [34958]
INFO:     Finished server process [34957]
INFO:     Stopping parent process [34952]

Answer 1

I found that it may be a better idea for me to start a BackgroundScheduler instead of an AsyncIOScheduler, and start it in the main function instead so the scheduler is configured in the parent process. Only 1 scheduler will be created in this case.

from fastapi import FastAPI
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
import uvicorn

def test():
    print(f"Test scheduler {datetime.now()}")
    
app = FastAPI()

if __name__ == "__main__":
    scheduler = BackgroundScheduler() 
    
    scheduler.add_job(test, trigger="cron", second="0-30")
    scheduler.start()

    uvicorn.run("main:app", workers=2)

    scheduler.shutdown()