Empresas
Empleos
  • Sobre nosotros
  • Soluciones
    • Publicación de vacantes
      Publica tu vacante y recibe candidatos calificados en 48h.
    • Evaluación de candidatos
      500+ pruebas técnicas y psicológicas, más anti-fraude.
    • Headhunting
      Búsqueda ejecutiva a la medida de principio a fin.
    • Nómina + EOR
      Dispersión de nómina y EOR en más de 15 países de LATAM.
  • Precios
  • Empleos

0

220
Vistas
Crear un activador de base de datos en un detector de eventos

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'
over 4 years ago · Santiago Trujillo
1 Respuestas
Responde la pregunta

0

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.

over 4 years ago · Santiago Trujillo Denunciar
Responde la pregunta
Encuentra empleos remotos

¡Descubre la nueva forma de encontrar empleo!

Top de empleos
Top categorías de empleo
Empresas
Publicar vacante Precios Comercial
Legal
Términos y condiciones Política de privacidad
© 2026 PeakU Inc. All Rights Reserved.
Andres GPT
Recomiéndame algunas ofertas
Necesito ayuda