我有一个简单的 Flask-SQLAlchemy 模型(带有创建触发器的事件监听器):
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
class Confirm(db.Model):
created = db.Column(db.DateTime, default=db.func.current_timestamp(), nullable=False)
modified = db.Column(db.DateTime, default=db.func.current_timestamp(), onupdate=db.func.current_timestamp(), nullable=False)
id = db.Column(db.String(36), primary_key=True)
class ConfirmOld(db.Model):
orig_created = db.Column(db.DateTime)
orig_modified = db.Column(db.DateTime)
orig_id = db.Column(db.String(36))
confirm_delete = DDL('''\
CREATE TRIGGER confirm_delete
BEFORE DELETE
ON confirm FOR EACH ROW
BEGIN
INSERT INTO confirm_old ( orig_created, orig_modified, orig_id )
VALUES ( OLD.created, OLD.modified, OLD.id );
END;
''')
event.listen(Confirm.__table__, 'after_create', confirm_delete)
当我运行 Alembic 迁移和升级时,未创建 TRIGGER(在 MySQL 中)。但是,它是在我使用 db.create_all() 时创建并正常工作。
是否可以让 Alembic/Flask-Migrate 来创建和管理我的触发器(即在 after_create 事件上运行的自定义 DDL)?
最佳答案
我遇到了同样的问题,尝试了使用 Replacable 对象的解决方案但没有奏效:
我设法通过编辑迁移脚本并执行触发器创建查询使其工作。
步骤如下:
运行
flask db migrate -m 'adding custom trigger on table x
它将在迁移文件夹的版本子文件夹下为您生成一个迁移脚本。检查在版本下创建的文件夹并像这样编辑它:
像这样创建你的触发器查询:
在文件中:
trigger = '''
CREATE TRIGGER confirm_delete
BEFORE DELETE
ON confirm FOR EACH ROW
BEGIN
INSERT INTO confirm_old ( orig_created, orig_modified, orig_id )
VALUES ( OLD.created, OLD.modified, OLD.id );
END;
'''
在升级方法中:
添加这一行:
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
# ### end Alembic commands ###
### add your queries here execute
op.execute(trigger)
如果您运行flask db upgrade
,它将执行查询并更新数据库
要降级数据库,请在降级方法中添加:
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
# ### end Alembic commands ###
op.execute('drop trigger if exists confirm_delete on confirm cascade;')
如果您检查您的数据库更改将被应用。
PS:更优雅的解决方案应该是建议的here
使用 Replaceable object,试过了但没用,可能是我的 alembic 没有更新。
解决方案应该是这样的:
创建一个 ReplaceableObjects 类:
class ReplaceableObject(object):
def __init__(self, name, sqltext):
self.name = name
self.sqltext = sqltext
用您的查询语句实例化它。
delete_trigger = ReplaceableObject('delete_trigger', trigger)
像这样更新升级和降级函数:
def upgrade():
op.create_sp(delete_trigger)
def downgrade():
op.drop_sp(delete_trigger)
希望对其他人有帮助...
关于flask - Alembic/Flask-Migrate 未检测到 after_create 事件,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/36455599/