7

我正在为 SQLAlchemy 开发一个 Audit mixin，但不确定如何执行此操作。

我的课看起来像这样：

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)

更新的很好，因为我只需要在表级别记录最新的更新；任何重要的审计都将保存在一个单独的表格中，详细说明更新/删除等。

我的问题是；我不希望更新 created_dt/by 列。我知道，在我的代码中，我可以在更新对象时简单地省略它们；但另一个编码器可能；所以我真的想确保在每次更新之前它会用自身覆盖值，或者如果有人试图更改它会引发错误（后者是首选）。

我的 SQLAlchemy 技能仍在开发中，是否可以解决事件？或者是否可以通过覆盖一些通用的声明性函数（如 save() 或 before_save() 或任何可能存在的函数来完成？

我会继续寻找答案 - 但最好是帮助找到解决方案（我不希望得到代码）。

4

2 回答 2

8

您的问题是您没有使用“默认”和“onupdate”的可调用对象。它记录在这里。

对于日期，它应该是（注意没有括号）：

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

或用户名：

default=lambda: current_user.username

后一个示例应该是一个函数而不是 lambda，以对 current_user 进行各种安全检查（例如，如果匿名呢？）

于 2013-11-07T15:43:14.717 回答
6

[编辑]我使用的是flask.g - 但我意识到除非在某处硬编码，否则它不会持久；因此，我在实际实现中转移到了用户会话 [/Edit]

好的，伙计们...希望这会对某人有所帮助。我想我已经整理出解决方案，经过测试并且对于我的项目来说足够安全（尽管它并不完美，我希望得到一些反馈）：

这是审计混合：

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

然后我只需将扩展和基本扩展添加到任何需要它们的模型上：

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

现在，我注意到这种方法有两个注意事项： - g.username 是硬编码的 - 在这个阶段，我不知道如何通过 SQLAlchemy 传递额外的参数以在 Mapper 中使用，所以现在；这将不得不做。- 仍然可以进行数据库级别的操作...在数据库上运行原始 SQL 不会阻止这些列的更新。

这些警告中的第二个是有问题的——但我认为定义模型级别的触发器可能有助于防止任何数据库交互问题。

关于不同方法的任何想法？我想我已经用这个锁定了任何基于 SQLalchemy 的小提琴......

于 2013-07-02T15:31:03.013 回答