Create a trigger in MySQL in SQLAlchemy migration with op.execute.
I wanted to have an update in last_update field, every time a user or an admin changed the data directly in the database (without python sqlalchemy). Instead of using signal, I created trigger inside op.execute method
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('new_table',
sa.Column('no_id', sa.Integer(), nullable=False),
sa.Column('car', sa.Integer(), nullable=False),
sa.Column('house', sa.Integer(), nullable=False),
sa.Column('tier_id', sa.Integer(), nullable=False),
sa.Column('last_updated', sa.TIMESTAMP(), nullable=True),
sa.Column('timestamp', sa.TIMESTAMP(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=True),
sa.ForeignKeyConstraint(['no_id'], ['companies.company_id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['tier_id'], ['tiers.id'], ondelete='CASCADE'),
sa.PrimaryKeyConstraint('no_id')
)
op.execute('''
CREATE TRIGGER tier_after_insert_trigger
BEFORE UPDATE ON tier_table
FOR EACH ROW
BEGIN
SET NEW.last_updated = CURRENT_TIMESTAMP();
END;
''')
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.execute('DROP TRIGGER IF EXISTS tier_after_insert_trigger')
op.drop_table('new_table')
Top comments (0)