在 Python 中使用 Peewee 在 SQL DDL 中使用 SQLite 触发器和日期时间默认值

Question

我有一个 SQLite 表，定义如下：

create table if not exists KeyValuePair (
    key         CHAR(255) primary key not null,
    val         text not null,
    fup         timestamp default current_timestamp not null,  -- time of first upload
    lup         timestamp default current_timestamp not null  -- time of last upload
);

create trigger if not exists entry_first_insert after insert
on KeyValuePair
begin
    update KeyValuePair set lup = current_timestamp where key = new.key;
end;

create trigger if not exists entry_last_updated after update of value
on KeyValuePair
begin
    update KeyValuePair set lup = current_timestamp where key = old.key;
end;

peewee.Model我正在尝试用 Python 为这个表编写一个。这是我到目前为止所拥有的：

import peewee as pw


db = pw.SqliteDatabase('dhm.db')
class BaseModel(pw.Model):
    class Meta:
        database = db

class KeyValuePair(BaseModel):
    key = pw.FixedCharField(primary_key=True, max_length=255)
    val = pw.TextField(null=False)
    fup = pw.DateTimeField(
        verbose_name='first_updated', null=False, default=datetime.datetime.now)
    lup = pw.DateTimeField(
        verbose_name='last_updated', null=False, default=datetime.datetime.now)

db.connect()
db.create_tables([KeyValuePair])

当我检查最后一行生成的 SQL 时，我得到：

CREATE TABLE "keyvaluepair" (
    "key" CHAR(255) NOT NULL PRIMARY KEY,
    "val" TEXT NOT NULL,
    "fup" DATETIME NOT NULL,
    "lup" DATETIME NOT NULL
);

所以我现在有两个问题：

1. entry_first_insert我一直无法找到一种方法来实现和触发器的行为entry_last_updated。peewee支持触发器吗？如果没有，有没有办法只从 .sql 文件而不是类定义创建表Model？
2. 有没有办法为 SQL 定义设置默认值fup并lup传播到 SQL 定义？

Answer 1

我已经找到了这两个问题的正确答案。该解决方案实际上在 SQL DDL 中强制执行所需的触发器和默认时间戳。

首先，我们定义一个便利类来包装触发器的 SQL。有一种更合适的方法可以对peewee.Node对象执行此操作，但我没有时间深入研究该项目的所有内容。此类Trigger仅提供字符串格式以输出用于触发器创建的正确 sql。

class Trigger(object):
    """Trigger template wrapper for use with peewee ORM."""

    _template = """
    {create} {name} {when} {trigger_op}
    on {tablename}
    begin
        {op} {tablename} {sql} where {pk} = {old_new}.{pk};
    end;
    """

    def __init__(self, table, name, when, trigger_op, op, sql, safe=True):
        self.create = 'create trigger' + (' if not exists' if safe else '')
        self.tablename = table._meta.name
        self.pk = table._meta.primary_key.name
        self.name = name
        self.when = when
        self.trigger_op = trigger_op
        self.op = op
        self.sql = sql
        self.old_new = 'new' if trigger_op.lower() == 'insert' else 'old'

    def __str__(self):
        return self._template.format(**self.__dict__)

接下来我们定义一个TriggerTable继承自BaseModel. 此类覆盖默认值，create_table以遵循表创建和触发器创建。如果任何触发器创建失败，则回滚整个创建。

class TriggerTable(BaseModel):
    """Table with triggers."""

    @classmethod
    def triggers(cls):
        """Return an iterable of `Trigger` objects to create upon table creation."""
        return tuple()

    @classmethod
    def new_trigger(cls, name, when, trigger_op, op, sql):
        """Create a new trigger for this class's table."""
        return Trigger(cls, name, when, trigger_op, op, sql)

    @classmethod
    def create_table(cls, fail_silently=False):
        """Create this table in the underlying database."""
        super(TriggerTable, cls).create_table(fail_silently)
        for trigger in cls.triggers():
            try:
                cls._meta.database.execute_sql(str(trigger))
            except:
                cls._meta.database.drop_table(cls, fail_silently)
                raise

下一步是创建一个类BetterDateTimeField。如果实例变量设置为函数，则此Field对象将覆盖默认值__ddl__以附加“DEFAULT current_timestamp”字符串。当然有更好的方法可以做到这一点，但这个方法捕获了基本用例。defaultdatetime.datetime.now

class BetterDateTimeField(pw.DateTimeField):
    """Propogate defaults to database layer."""

    def __ddl__(self, column_type):
        """Return a list of Node instances that defines the column."""
        ddl = super(BetterDateTimeField, self).__ddl__(column_type)
        if self.default == datetime.datetime.now:
            ddl.append(pw.SQL('DEFAULT current_timestamp'))
        return ddl

最后，我们定义新的和改进的KeyValuePair模型，结合我们的触发器和日期时间字段改进。我们通过创建表来结束 Python 代码。

class KeyValuePair(TriggerTable):
    """DurableHashMap entries are key-value pairs."""

    key = pw.FixedCharField(primary_key=True, max_length=255)
    val = pw.TextField(null=False)
    fup = BetterDateTimeField(
        verbose_name='first_updated', null=False, default=datetime.datetime.now)
    lup = BetterDateTimeField(
        verbose_name='last_updated', null=False, default=datetime.datetime.now)

    @classmethod
    def triggers(cls):
        return (
            cls.new_trigger(
                'kvp_first_insert', 'after', 'insert', 'update',
                'set lup = current_timestamp'),
            cls.new_trigger(
                'kvp_last_udpated', 'after', 'update', 'update',
                'set lup = current_timestamp')
        )

KeyValuePair.create_table()

现在架构已正确创建：

sqlite> .schema keyvaluepair
CREATE TABLE "keyvaluepair" ("key" CHAR(255) NOT NULL PRIMARY KEY, "val" TEXT NOT NULL, "fup" DATETIME NOT NULL DEFAULT current_timestamp, "lup" DATETIME NOT NULL DEFAULT current_timestamp);
CREATE TRIGGER kvp_first_insert after insert
    on keyvaluepair
    begin
        update keyvaluepair set lup = current_timestamp where key = new.key;
    end;
CREATE TRIGGER kvp_last_udpated after update
    on keyvaluepair
    begin
        update keyvaluepair set lup = current_timestamp where key = old.key;
    end;
sqlite> insert into keyvaluepair (key, val) values ('test', 'test-value');
sqlite> select * from keyvaluepair;
test|test-value|2015-12-07 21:58:05|2015-12-07 21:58:05
sqlite> update keyvaluepair set val = 'test-value-two' where key = 'test';
sqlite> select * from keyvaluepair;
test|test-value-two|2015-12-07 21:58:05|2015-12-07 21:58:22