Tengo una API de matraz y estoy usando Flask-SQLAlchemy para manejar una base de datos SQLite. Tengo una tabla que almacena entradas de registro y quiero limitar el número máximo de filas a un número n. Dado que las inserciones también se realizan desde otro script fuera del matraz usando SQL sin procesar, creé un disparador que verifica la cantidad de filas y elimina las más antiguas si el número es mayor que 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 Este disparador funciona como se esperaba, pero estoy luchando para configurarlo usando matraz-sqlalchemy. ¿Cómo puedo configurar el activador/ejecutar SQL sin procesar usando matraz-sqlalchemy? El SQL solo debe ejecutarse una vez después de la creación de la base de datos, por lo que intento ejecutarlo justo después de mi instrucción create_all(). Me topé con esta respuesta de StackOverflow que sugirió una solución que aparentemente quedará obsoleta pronto. También leí la documentación de SQLAlchemy sobre DDL personalizado , pero no sé cómo crear este DDL personalizado con matraz_sqlalchemy. Cuando creo el DDL como en la documentación de SQLAlchemy, aparece un mensaje de error que dice que 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())Mi modelo se define usando el modelo base declarativo de Flass-sqlalchemy:
class ConnectionLog(db.Model): __tablename__ = 'connection_logs'No necesita crear una instancia DDL , puede ejecutar el SQL dentro de la función de escucha. Los documentos relevantes están aquí .
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)Asegúrese de que el intérprete haya leído este código antes de crear las tablas.