Automating Data pulling with Python using the APScheduler library

Pulling data from an API Endpoint and manually updating said data can be a pain, but you can schedule this instead.

Author

Carson Szeder
February 08, 2024

When using Python’s AP Scheduler library, you can automate data pulling every X seconds/hours/days, and it’s really simple (Under 30 lines of code).

Let’s say you’re using a FastAPI Set up. you’re code would look like this at the bottom:

def call_endpoint(): response = requests.get("http://127.0.0.1:8000/api/get_dummy_data") print(response.json()) #Insert data base update logic scheduler = AsyncIOScheduler() @app.on_event("startup") async def startup_event(): scheduler.add_job(call_endpoint, 'interval', seconds=10) scheduler.start()

Before setting this up, obviously you need the APScheduler library installed.

our call_endpoint function does one thing: Hit an endpoint and print the data. In the case above I used a mock endpoint I created, you can replace with whatever endpoint you want to pull from. Then where I put #Insert data base update logic, you would add what you want to do with that data after.

@app.on_event("startup") async def startup_event(): scheduler.add_job(call_endpoint, 'interval', seconds=10) scheduler.start()

This code above runs the scheduler when the API Itself is ran. it’ll add the job and you can select what interval/what function you want to run. In my case I did 10 seconds.

Hope you found this helpful!