Is it possible to create trigger on all tables by modifying declarative_base?

[hidetatz]

I want to add updated_at column on every table I manage using alembic. Because we are using PostgreSQL, the automatic update on the updated_at column requires trigger.

Trigger will look like this:

    CREATE TRIGGER tg_auto_update_updated_at
    BEFORE UPDATE ON "{table name here}"
    FOR EACH ROW
    EXECUTE FUNCTION fn_update_updated_at(); -- please regard this function already exists

Because the column updated_at and the trigger should exist in every table, I don't want to add them in all tables manually but want to automate this by modifying the declarative_base class which is inherited by every model class.

Base = declarative_base(metadata=meta)
Base.updated_at = Column(DateTime(timezone=True)) # add updated_at column in all the tables

class User(Base):
     # other columns definitions

I made sure the above works, the updated_at is added in tables.
I want to do a similar thing with the trigger. Can anyone tell how it can be done?

[zzzeek]

Are you looking for trigger DDL() to be added to your model as you have above, or within your actual alembic migrations? There are different approaches you'd be taking. if you are looking for alembic migrations to automatically include additional steps that aren't supported by autogenerate you'd be using process_revision_directives and you'd be rendering additional ExecuteSQLOp directives into the structure.

[hidetatz]

Hi, thank you for your answer.

within your actual alembic migrations?

I want to include trigger creation in alembic migration.

Let me clarify what I exactly want to do. Now, I have base class and model classes which inherits it like this.

# base class
Base = declarative_base(metadata=meta)

# model classes
class User(Base):
    __tablename__ = "..."
    id = Column(...)

class Product(Base):
    __tablename__ = "..."
    id = Column(...)

# ... a lot of other tables

Now, I want to create updated_at column in all the tables and update the column automatically on record update.
To achieve the latter one, automatic column update, because I am using PostgreSQL, I need to use trigger like this.

   CREATE TRIGGER tg_auto_update_updated_at
   BEFORE UPDATE ON "user"
   FOR EACH ROW
   EXECUTE FUNCTION fn_update_updated_at();

Because model classes (tables) will be newly created in the future also, I want to make below thing work:

1. I want to create updated_at column in all the existing table without manually adding it into each model class.
2. I want to make sure updated_at column is automatically created when a new model class is defined in the future.
3. I want to create trigger in all the existing table without manually adding it into each model class.
4. I want to make sure the trigger is automatically created when a new model class is defined in the future.

For 1 and 2, I already achieved them. I made Base.updated_at field like this, and because all the model classes inherit Base, the updated_at column is created. I made sure running alembic revision --autogenerate creates a migration file which includes op.add_column operation on all the tables.

Base.updated_at = Column(DateTime(timezone=True))

So now I want to achieve 3 and 4. Because I want to make sure the trigger is created in newly created model classes in the future, I'm trying to modify our Base for that purpose like what I did for updated_at.

I found that alembic_utils allows me to manage PGTrigger by migration files. However, this PGTrigger requires the concrete table name to be created. like this:

tg_update_system_updated_at = PGTrigger(
    schema="public",
    signature="tg_update_system_updated_at",
    on_entity="public.user",
    definition="""
        BEFORE UPDATE ON "user"
        FOR EACH ROW EXECUTE FUNCTION function_update_system_updated_at();
    """,
)

So because I want to ensure the trigger is all the table including newly created one in the future, I don't want to specify concrete table name in the PGTrigger.
So I was thinking if the Base can be used for this purpose.

Now, I want to get back to your answer. In my understanding, I can use process_revision_directives and ExecuteSQLOp like this:

# add this function
def process_revision_directives(context, revision, directives):
    script = directives[0]
    for op in script.upgrade_ops:
        if type(op) == CreateTableOp:
            # create trigger on newly created tables
            script.upgrade_ops.append(ExecuteSQLOp(f"CREATE TRIGGER tg_auto_update_updated_at BEFORE UPDATE ON "{op.table_name}" FOR EACH ROW EXECUTE FUNCTION fn_update_updated_at();"))

# ...

def run_migrations_online():

    # ...
    with engine.connect() as connection:

        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            process_revision_directives=process_revision_directives) # add this

        with context.begin_transaction():
            context.run_migrations()

Is my understanding correct?

[zzzeek]

So I was thinking if the Base can be used for this purpose.

SQLAlchemy has no notion of a "trigger" object directly, and the feature you are using with Base is specific to ORM mapped attributes that map to database columns. So there is no facility in Alembic autogenerate to read records from a Base class with a mapped table and then generate "triggers".

alembic_utils does have an explicit "trigger" notion but I am not familiar with the extent to which it has produced an "autogenerate" feature which would work from some metadata-level directives to produce them. if it did, you still could not use Base.column = Column() because that feature only applies to mapped columns; you instead would use an event like before_parent_attach to intercept when new Table objects are being associated with their Metadata collection, and at that point generate whatever kind of "trigger" object alembic_utils provides, which would automate the production of "alembic_utils trigger object (hypothetical, since I dont know) per class". Then autogenerate, augmented by alembic_utils, would generate this PGTrigger migration. This is all stuff you'd probably want to ask of alembic_utils authors.

Now, I want to get back to your answer. In my understanding, I can use process_revision_directives and ExecuteSQLOp like this:

yes, your example is the basic thing I was proposing (which you can see is not that complicated). you can just have the direct SQL you want to run each time.

[hidetatz]

Understood that the alembic Base class concept does not match with trigger. I also made sure the example you provided works as expected, so will follow it. Thank you for your help!

[stephanschielke]

I didn't manage to let Alembic generate the migration but you can manually add the following to your migration script:

def upgrade() -> None:
    op.execute(
        text("""
        CREATE OR REPLACE FUNCTION update_updated_at()
            RETURNS TRIGGER AS
        $$
        BEGIN
            NEW.updated_at = now();
            RETURN NEW;
        END;
        $$ language 'plpgsql';
        """)
    )
    
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column("table_name", sa.Column("updated_at", sa.DateTime(), nullable=True))
    ...
    # ### end Alembic commands ###

    # Get all tables with an updated_at column
    tables = sa.inspect(op.get_bind()).get_table_names()
    for table in tables:
        if sa.inspect(op.get_bind()).get_columns(table):
            columns = [c["name"] for c in sa.inspect(op.get_bind()).get_columns(table)]
            if "updated_at" in columns:
                # Create the trigger for the table
                op.execute(f"""
                    CREATE TRIGGER update_{table}_updated_at_trigger
                    BEFORE UPDATE ON {table}
                    FOR EACH ROW
                    EXECUTE PROCEDURE update_updated_at();
                """)

For the downgrade:

def downgrade() -> None:
    tables = sa.inspect(op.get_bind()).get_table_names()
    for table in tables:
        if sa.inspect(op.get_bind()).get_columns(table):
            columns = [c["name"] for c in sa.inspect(op.get_bind()).get_columns(table)]
            if "updated_at" in columns:
                # Drop the trigger for the table
                op.execute(f"""
                    DROP TRIGGER IF EXISTS update_{table}_updated_at_trigger ON {table};
                """)

    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_column("table_name", "updated_at")
    ...
    # ### end Alembic commands ###

    # Drop the function
    op.execute("""
        DROP FUNCTION IF EXISTS update_updated_at();
    """)