onupdate not overridinig current datetime value

Question

I'm working on a Audit mixin for SQLAlchemy and not sure exactly how to do this.

My class looks like this:

class AuditColumns(object):

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

    created_by = Column(String(64),
                        default=current_user,
                        nullable=False)

    updated_dt = Column(DateTime,
                        default=datetime.utcnow(),
                        nullable=False,
                        onupdate=datetime.utcnow())
    updated_by = Column(String(64),
                        default=current_user,
                        nullable=False,
                        onupdate=current_user)

The updated ones are fine as I only need to to record the latest update at the table level; any important audits will be held in a separate table that detail the update/delete etc.

My problem is; I don't want the created_dt/by columns to be updated. I know, in my code I can simply omit them when updating the objects; but another coder may; so I really want to either ensure before every update it overrides the value with itself, OR raises an error if someone tries to change it (the latter is preferred).

My SQLAlchemy skills are still in development, would events be a place to figure out; or is there something that can be done by override some generic declarative function like save() or before_save() or whatever may exist?

I will keep looking for the answer - but help finding the solution (I would prefer not to be given code) is preferable.

Answer 1

Your problem is that you are not using callables for "default" and "onupdate". It's documented here.

for dates it should be (notice the absence of parentheses):

default=datetime.datetime.utcnow
onupdate=datetime.datetime.utcnow

or for username:

default=lambda: current_user.username

The latter example should be a function rather than a lambda, to make various safety checks against current_user (what if anonymous for example?)

Answer 2

[Edit] I was using flask.g - but I've realised it's not persistent unless hardcoded somewhere; hence I've moved to user sessions in my real implementations [/Edit]

OK guys... Hopefully this will help someone. I think I've sorted out the solution, tested and is secure enough for my project (although it's not perfect and I would love some feedback):

Here is the Audit Mixin:

from datetime import datetime
from flask import g
from sqlalchemy import Column, DateTime, String
from sqlalchemy.orm import MapperExtension


class AuditColumns(object):

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

    created_by = Column(String(64),
                        nullable=False)

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

    updated_by = Column(String(64),
                        nullable=False)


class AuditExtension(MapperExtension):

    def before_insert(self, mapper, connection, instance):
        """ Make sure the audit fields are set correctly  """
        instance.created_dt = datetime.utcnow()
        instance.created_by = g.username

        instance.updated_dt = datetime.utcnow()
        instance.updated_by = g.username

    def before_update(self, mapper, connection, instance):
        """ Make sure when we update this record the created fields stay unchanged!  """
        instance.created_dt = instance.created_dt
        instance.created_by = instance.created_by

        instance.updated_dt = datetime.utcnow()
        instance.updated_by = g.username

Then I simply tack the extension and the base extension onto whatever models need them:

class Roles(db.Model, AuditColumns):

    id = Column(BigInteger, primary_key=True)
    username = Column(String(64), nullable=False, unique=True)
    password = Column(String(255), nullable=False)

    __mapper_args__ = {
        'extension': AuditExtension()}

    def __repr__(self):
        return self.username

Now, there are 2 caveats I've noticed to this approach: - g.username is hardcoded - at this stage I have no idea how I would pass additional args through SQLAlchemy to be used inside the Mapper, so for now; this will have to do. - Database level manipulation can still be done... Running raw SQL on the database won't prevent these columns from being updated.

The second of these caveats is problematic - but I think defining model level triggers may help prevent any db interaction fiddles.

Any ideas on perhaps a different approach? I think I've locked up any SQLalchemy based fiddles with this...