Poll SQL table using Python?

Question

I want to implement something similar to this, where the custom function classify_process() continuously asks a Redis db for new entries.

Can this be done a regular SQL table? The db.pipeline() seems to be Redis-specific, so I'm not sure how to imitate this functionality on a SQL db.

My goal is to check one table for new rows, and if there's a new row, run a prediction using a ML model.

QUESTION: How can I continously check if there are any new records in a MS SQL table (which has an interger ID column that automatically increments) and then trigger a function?

My idea of what has to happen using sqlalchemy:

import time 
from sqlalchemy import create_engine

from sqlalchemy.orm import sessionmaker

DRIVER = 'driver'
SERVER = 'server'
DATABASE = 'database'


engine = create_engine(f'mssql+pyodbc://{SERVER}/{DATABASE}?{DRIVER}')

SessionLocal = sessionmaker(bind=engine)

def some_process():
    while True:

        # look for new rows in table, then execute prediction on newest rows
        ...some code...?

        # Sleep for x seconds

        time.sleep(5)


if __name__ == "__main__":
    some_process()

user_2144390 · Answer

With an auto-incrementing integer (IDENTITY(1,1)) primary key you would check for new rows by looking up the largest PK value …

SELECT TOP 1 id FROM your_table ORDER BY id DESC

… and compare it with the previous largest value, like this

max_id_query = session.query(Thing.id).order_by(Thing.id.desc()).limit(1)
last_max_id = max_id_query.scalar()
while True:
    max_id = max_id_query.scalar()
    if max_id > last_max_id:
        print(
            f"New row(s) found. "
            f"Processing ids {last_max_id + 1} through {max_id}"
        )
        # … do stuff

        last_max_id = max_id
    time.sleep(5)

Poll SQL table using Python?

1 Answers