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 Scheduletry: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

Love podcasts or audiobooks? Learn on the go with our new app.

Recommended from Medium

An incremental approach to a database migration, or how throughput and latency are not the same…

Cloudflare — Get Caught in the NET

How To: Manage Docker & Kubernetes concurrently with Portainer in Docker Desktop

TerraMaster D5–300 Raid Direct Attached Storage review | Vic B’Stard’s State of Play

How to Enable HTTP Requests Roblox

Setting up Nginx configuration to get an ‘A+’ in SSL Labs Server Test

A leap forward for STAC: SpatioTemporal Asset Catalog 0.6.0 Specification Released!

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store
Andrei Hawke

Andrei Hawke

More from Medium

Deploy a containerised fastapi APP in minikube

muzz goes open source

DEPLOYING DJANGO APP TO GOOGLE CLOUD USING CLOUD RUN

Connect to Docker Service using hostname in Docker Swarm