CODEWITHSUNDEEP
pythonsqlalchemy
Trent
Trent

Reputation: 3103

SQLAlchemy events after_create for all models

I'm working on a project that requires generic customisation to every model that is created. The way I have done most of the work to date is by model inheritance. Here is my code block to give you a better idea:

app.core.dba.mixins:

class AuditExtension(MapperExtension):
    """
    AuditExtension enforces the audit column values, and ensures any interaction with
    SQLAlchemy cannot override the values
    """

    def before_insert(self, mapper, connection, instance):
        instance.created_dt = datetime.utcnow()
        instance.created_by = audit_session_user()

        instance.updated_dt = datetime.utcnow()
        instance.updated_by = audit_session_user()

    def before_update(self, mapper, connection, instance):
        # Never update the created columns
        instance.created_dt = instance.created_dt
        instance.created_by = instance.created_by

        instance.updated_dt = datetime.utcnow()
        instance.updated_by = audit_session_user()


class AuditColumns(object):

    """ Generate the column schema for simple table level auditing. """
    created_dt = Column(DateTime,
                        default=datetime.utcnow(),
                        nullable=False)
    created_by = Column(String(64),
                        #ForeignKey('operators.username', ondelete="RESTRICT"),
                        nullable=False)

    updated_dt = Column(DateTime,
                        default=datetime.utcnow(),
                        nullable=False,
                        onupdate=datetime.utcnow())

    updated_by = Column(String(64),
                        #ForeignKey('operators.username', ondelete="RESTRICT"),
                        nullable=False)

    __mapper_args__ = {
        'extension': AuditExtension()}

My models then inherit AuditColumns:

class ObjectTypes(Base, AuditColumns):
    __tablename__ = 'object_types'

    id = Column(BigInteger, primary_key=True)
    name = Column(String, nullable=False, unique=True)

    def __repr__(self):
        return self.name

My problem is; my solution for forcing the audit data works as long as operation is contained to the flask app and SQLAlchemy - this does not prevent anyone with database access to updating the values.

Therefore, I now need to implement a trigger on each model that inherits the AuditColumns. I found this post Sqlalchemy mixins / and event listener - and it describes a method for before_insert/update (which I've previous had working), but not for "after_create".

Now, I've added this to my mixins file code (straight after my audit code above:

trig_ddl = DDL("""
            CREATE TRIGGER tr_audit_columns BEFORE INSERT OR UPDATE
            ON test_table
            FOR EACH ROW EXECUTE PROCEDURE
            ss_test();
        """)

event.listen(AuditColumns, 'after_create', trig_ddl)

However, when I run a test case:

Base.metadata.drop_all(db.get_engine(app))
Base.metadata.create_all(db.get_engine(app))

I get the following error:

File "D:\Devel\flask-projects\sc2\app\core\dba\mixins.py", line 59, in <module>
    event.listen(AuditColumns, 'after_create', trig_ddl)
  File "D:\Devel\flask-projects\env\lib\site-packages\sqlalchemy\event.py", line 43, in listen
    (identifier, target))
sqlalchemy.exc.InvalidRequestError: No such event 'after_create' for target '<class 'app.core.dba.mixins.AuditColumns'>'

I'm guessing this is because it's not a table yet; but how would I globally define an event listener for a table create, that will execute this type of command?

I know I have to make the trig_ddl dynamic (which I don't think will be too hard, but I at least need to figure out the global element of this).

Basically, I don't want people to have to manually write in each model this event when it's clearly tied to these audit columns.

Any push in the right direction would be great.

Upvotes: 5

Views: 7595

Answers (1)

zzzeek
zzzeek

Reputation: 75317

Well you'd need to piggyback the events here so you can get at that Table:

@event.listens_for(AuditColumns, "instrument_class", propagate=True)
def instrument_class(mapper, class_):
    if mapper.local_table is not None:
        trigger_for_table(mapper.local_table)

def trigger_for_table(table):
    trig_ddl = DDL("""
                CREATE TRIGGER tr_%s_audit_columns BEFORE INSERT OR UPDATE
                ON %s
                FOR EACH ROW EXECUTE PROCEDURE
                ss_test();
            """ % (table.name, table.name))

    event.listen(table, 'after_create', trig_ddl)

any subclass of AuditColumms is mapped, mapper.local_table will be there already (as will class.__table__, same thing), you apply the DDL event at that point.

Upvotes: 7

Related Questions