Scheduled Jobs with FastAPI and APScheduler

https://unsplash.com/photos/LXWUK-gypVc

The Challenge:

• Show how to use APScheduler to schedule ongoing Jobs
• Use a practical example
• Adhere to good FastAPI principles (such as Pydantic Models)
• Provide Some Smarts around scheduling
• Provide a reusable codebase for others to build on
• Identify gaps / room for improvement

The Use Case:

• SSL Certificate checks are a useful thing to do from a monitoring point of view as expired certificates can cause a whole lot of unexpected issues / outages.
• There are plenty of online services available to do SSL Monitors however we might have infrastructure behind firewalls or may not want to expose our domains to public checks.
• We are going to build a small web service to schedule and execute SSL Checks based on FastAPI.

Building Blocks:

Models:

https://unsplash.com/photos/4nulm-JUYFo

Made ya look, no that’s not the models I am talking about.

Pydantic makes it easy enough to model responses and this is useful to ensure that we build a model once and write code that adheres to that model

We need 3x models

• A) Model for Certificate Check Results
• B) A model for Responding to Job Creation / Deletion Requests
• C) A model to represent the current Jobs Managed By the Scheduler

what our models actually look like

Endpoints:

• A) An Endpoint to test the SSL Checks (I have added google.com as a default to demonstrate)
• B) An endpoint to schedule a job
• C) An endpoint to delete a job
• D) An endpoint to view current jobs
• E) Future State — and endpoint to update a job

This loosely follows the CRUD approach.

Persistence of the Schedules:

• I decided to utilise the jobschedules functionality and the FastAPI on startup event.
• I also declare the Schedule as a Global Variable to ensure other functions can update it and manipulate it

@app.on_event("startup")
async def load_schedule_or_create_blank():
global Schedule
try:
jobstores = {
'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
Schedule = AsyncIOScheduler(jobstores=jobstores)
Schedule.start()
logger.info("Created Schedule Object")
except:
logger.error("Unable to Create Schedule Object")

Running the Code

• We will need a few extra libraries to make this work.
• Below is the command to install the required dependencies and run the app (the same way you would any uvicorn app in the demos).

pip install cryptography APSCheduler SQLAlchemy
git clone https://github.com/cryptoroo/fastapi-scheduled-ssl-checks.git
cd fastapi-scheduled-ssl-checks 
uvicorn main:app

• The interface is fairly intuitive I think. When you schedule a job you should see logging showing that the jobs are running.

• We can view the current scheduled jobs by using the get endpoint

Conclusions

• APScheduler provides very powerful scheduling functionality natively
• FastAPI allows us to create APIs quickly and effectively. Giving us reliable and easy to use experience.
• Writing what may have be traditionally thought of as “difficult” scripts is made simpler.
• Having Schedulable APIs allows to remove the need for running CRON jobs one bit of our environment and scripts to check as parts of our pipelines.
• I didn’t quite manage to get the Update Functionality working but happy with being able to schedule jobs effectively