15

我有几个要在创建表后运行的自定义 DDL 语句：

update_function = DDL("""                                                                                                                                                       
CREATE OR REPLACE FUNCTION update_timestamp()
RETURNS TRIGGER AS $$
BEGIN
    NEW.updated_at = now();
    RETURN NEW;
END;
$$ language 'pgplsql';
""")

update_trigger = DDL("""
CREATE TRIGGER update %(table)s_timestamp BEFORE UPDATE
ON %(table)s FOR EACH ROW EXECUTE PROCEDURE update_timestamp();
""")

我已经像这样附加它们：

event.listen(Session.__table__, 'after_create', update_function)
event.listen(Session.__table__, 'after_create', update_trigger)

当我这样做时create_all，我得到了我期望的 SQL：

CREATE OR REPLACE FUNCTION update_timestamp()
RETURNS TRIGGER AS $$ 
BEGIN
    NEW.updated_at = now();
    RETURN NEW;
END;
$$ language 'pgplsql';


CREATE TRIGGER update session_timestamp BEFORE UPDATE
ON session FOR EACH ROW EXECUTE PROCEDURE update_timestamp();

但是当我使用 Alembic 升级时，语句不会出现：

-- Running upgrade c0d470e5c81 -> 6692fad7378

CREATE TABLE session (
    created_at TIMESTAMP WITHOUT TIME ZONE DEFAULT 'CURRENT_TIMESTAMP', 
    updated_at TIMESTAMP WITHOUT TIME ZONE DEFAULT 'CURRENT_TIMESTAMP', 
    id VARCHAR(32) NOT NULL, 
    owner_id INTEGER, 
    database_id VARCHAR(32), 
    content TEXT, 
    PRIMARY KEY (id), 
    FOREIGN KEY(database_id) REFERENCES database (id), 
    FOREIGN KEY(owner_id) REFERENCES users (id)
);

INSERT INTO alembic_version (version_num) VALUES ('6692fad7378');

有没有办法让 alembic 触发“after_create”事件？

4

3 回答 3

12

The table level before_create/after_create events are emitted (just not the metadata level ones). you need to make sure whatever happens in your env.py script ultimately involves those event listeners being set up.

The code you have here looks a little suspect:

event.listen(Session.__table__, 'after_create', update_function)
event.listen(Session.__table__, 'after_create', update_trigger)

Session.__table__ here would just be a single Table instance and that's probably not what you'd see in the alembic script. The alembic create_table command creates a Table locally and just runs a create on it, so you'd need to listen to all Table objects globally:

from sqlalchemy import Table
event.listen(Table, 'after_create', update_function)
event.listen(Table, 'after_create', update_trigger)

if these events are only for this one specific table though, then you wouldn't use any events, you'd just put the DDL() for those triggers directly in your migration script, right after where it calls create_table().

于 2013-06-03T23:54:36.097 回答
5

扩展@zzzeek 的答案，这个助手对我有用：

from sqlalchemy import Table
from sqlalchemy.event import listen
from functools import partial

def on_table_create(class_, ddl):

    def listener(tablename, ddl, table, bind, **kw):
        if table.name == tablename:
            ddl(table, bind, **kw)

    listen(Table,
           'after_create',
           partial(listener, class_.__table__.name, ddl))

然后你会这样做：

on_table_create(Session, update_function)
on_table_create(Session, update_trigger)
于 2015-12-01T20:10:12.483 回答
1

这个解决方案对我帮助很大：https ://gist.github.com/jasco/5f742709088f80f07eb2e0d6a141d3f2

因此，如果您有一个带有侦听器的 DDL，请将其放入变量中：

create_table_ddl = DDL("...").execute_if(dialect='postgresql')
drop_table_ddl = DDL("...").execute_if(dialect='postgresql')


event.listen(Model.__table__, 'after_create', create_table_ddl)
event.listen(Model.__table__, 'before_drop', drop_table_ddl)

在迁移中手动执行此操作：

def upgrade():
    create_table_ddl(target=None, bind=op.get_bind())

def upgrade():
    drop_table_ddl(target=None, bind=op.get_bind())

Alembic 不会调用您的表事件

于 2021-02-03T21:35:29.103 回答