如何在Flask-SQLAlchemy中通过事件监听器创建SQLite数据库触发器

问题原因 #

你的报错和配置错误来自两个核心问题：

• 注册事件监听时错误调用了trigger.execute()，你应该传入DDL对象本身而非直接执行它，此时还未建立数据库连接，直接执行自然会抛出未绑定引擎/连接的错误
• 拼接DDL字符串时没有预留空格，相邻字符串拼接后会出现语法错误，比如'trigger_log_insert'BEFORE INSERT这类非法语句

方案1：使用DDL事件自动创建（表创建时自动触发） #

首先调整你的触发器定义和事件监听代码：

from sqlalchemy import DDL, event

# 优先用三引号写多行SQL，避免拼接空格问题
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
""")

# 监听表的after_create事件，传入DDL对象本身即可，不要加()执行
event.listen(ConnectionLog.__table__, 'after_create', trigger)

后续调用db.create_all()时，会自动在connection_logs表创建完成后执行触发器创建语句。

方案2：create_all后手动执行原生SQL（符合你只执行一次的需求） #

如果你只需要在首次建库时执行一次，不需要绑定表创建事件，可以直接在create_all()调用后执行原生SQL，逻辑更直观：

from sqlalchemy import text

# 先建表
db.create_all()

# 再执行触发器创建语句
create_trigger_sql = """
CREATE TRIGGER IF NOT EXISTS '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
"""
with db.engine.connect() as conn:
    conn.execute(text(create_trigger_sql))
    conn.commit()

加IF NOT EXISTS可以避免重复创建触发器报错，适配多次执行的场景。

内容的提问来源于stack exchange，提问作者Gasp0de