Create a database trigger in an event listener

Question

I have a flask API and I'm using Flask-SQLAlchemy to handle a SQLite database. I have a table which stores log entries, and I want to limit the maximum number of rows to a number n. Since insertions are also made from another script outside of flask using raw SQL, I created a trigger that checks the number of rows and deletes the oldest ones if the number is higher than n:

CREATE TRIGGER 'trigger_log_insert' 
BEFORE INSERT ON 'connection_logs' 
WHEN ( SELECT count(*) FROM  'connection_logs' ) > 5 
BEGIN 
DELETE FROM 'connection_logs' 
WHERE id NOT IN ( SELECT id FROM 'connection_logs' ORDER BY id DESC LIMIT 5 ); 
END

This trigger works as expected, but I am struggling to set it using flask-sqlalchemy. How can I set the trigger / execute raw SQL using flask-sqlalchemy? The SQL only needs to be executed once after db creation so I intent to execute it right after my create_all() statement. I stumbled upon this StackOverflow answer which suggested a solution that is apparently going to be deprecated soon. I also read the SQLAlchemy documentation about custom DDL, but I don't know how to create this custom DDL with flask_sqlalchemy. When I create the DDL like in the SQLAlchemy documentation, I get an error message saying DDL object is not bound to an Engine or Connection.:

trigger = DDL(
"CREATE TRIGGER 'trigger_log_insert'"
"BEFORE INSERT ON 'connection_logs'"
"WHEN ( SELECT count(*) FROM  'connection_logs' ) > 5"
"BEGIN"
"DELETE FROM 'connection_logs' WHERE id NOT IN"
"("
"SELECT id FROM 'connection_logs' ORDER BY id DESC LIMIT 5"
");"
"END"
)

event.listen(ConnectionLog, 'after_create', trigger.execute())

My model is defined using flask-sqlalchemy's declarative base model:

class ConnectionLog(db.Model):
    __tablename__ = 'connection_logs'

snakecharmerb · Accepted Answer · 2021-09-19 17:01:12Z

5

+50

You don't need to create a DDL instance, you can execute the SQL within the listener function. The relevant docs are here.

import sqlalchemy as sa
...

class ConnectionLog(db.Model):
    __tablename__ = 'connection_logs'
...

def after_create(target, connection, **kw):
    connection.execute(sa.text("""\
        CREATE TRIGGER 'trigger_log_insert'
        BEFORE INSERT ON 'connection_logs'
        WHEN ( SELECT count(*) FROM  'connection_logs' ) > 5
        BEGIN
        DELETE FROM 'connection_logs' WHERE id NOT IN
        (
        SELECT id FROM 'connection_logs' ORDER BY id DESC LIMIT 5
        );
        END
        """
    ))

# Listen on the underlying table object, not on the model class.
sa.event.listen(ConnectionLog.__table__, "after_create", after_create)

Ensure that the interpreter has read this code before creating the tables.

answered Sep 19, 2021 at 17:01

snakecharmerb

57.1k13 gold badges136 silver badges200 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Gasp0de Over a year ago

Works like a charm and in addition I also learned something about ORM events, thank you!

Collectives™ on Stack Overflow

Create a database trigger in an event listener

1 Answer 1

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related