如何在Postgres中使用CURRENT_TIMESTAMP设置SQLAlchemy的默认设置？
如何在Postgres中使用CURRENT_TIMESTAMP设置SQLAlchemy的默认设置？
提问于 2022-02-01 19:08:05
这是How to set DEFAULT ON UPDATE CURRENT_TIMESTAMP in mysql with sqlalchemy?的姐妹问题，但关注的是Postgres而不是MySQL。
假设我们想要创建一个包含列users的表datemodified，在默认情况下，每当行被更新时，它都会更新到当前时间戳。在MySQL姐妹PR中给出的解决方案是：
user = Table(
"users",
Metadata,
Column(
"datemodified",
TIMESTAMP,
server_default=text("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"),
),
)如何使用Postgres后端获得相同的功能？
Stack Overflow用户
回答已采纳
发布于 2022-02-10 21:15:20
最后，我按照a_horse_with_no_name在注释中的建议使用触发器实现了这一功能。完整的SQLAlchemy实现和与Alembic的集成遵循。
SQLAlchemy实现
# models.py
class User(Base):
__tablename__ = "user"
id = Column(Integer, primary_key=True)
name = Column(Text)
created_at = Column(DateTime, server_default=sqlalchemy.func.now(), nullable=False)
updated_at = Column(DateTime)# your_application_code.py
import sqlalchemy as sa
create_refresh_updated_at_func = """
CREATE FUNCTION {schema}.refresh_updated_at()
RETURNS TRIGGER
LANGUAGE plpgsql AS
$func$
BEGIN
NEW.updated_at := now();
RETURN NEW;
END
$func$;
"""
create_trigger = """
CREATE TRIGGER trig_{table}_updated BEFORE UPDATE ON {schema}.{table}
FOR EACH ROW EXECUTE PROCEDURE {schema}.refresh_updated_at();
"""
my_schema = "foo"
engine.execute(sa.text(create_refresh_updated_at_func.format(schema=my_schema)))
engine.execute(sa.text(create_trigger.format(schema=my_schema, table="user")))与Alembic的集成
在我的例子中，重要的是将触发器创建与Alembic集成起来，并将触发器添加到n个维度表(它们都有一个updated_at列)。
# alembic/versions/your_version.py
import sqlalchemy as sa
create_refresh_updated_at_func = """
CREATE FUNCTION {schema}.refresh_updated_at()
RETURNS TRIGGER
LANGUAGE plpgsql AS
$func$
BEGIN
NEW.updated_at := now();
RETURN NEW;
END
$func$;
"""
create_trigger = """
CREATE TRIGGER trig_{table}_updated BEFORE UPDATE ON {schema}.{table}
FOR EACH ROW EXECUTE PROCEDURE {schema}.refresh_updated_at();
"""
def upgrade():
op.create_table(..., schema="foo")
...
# Add updated_at triggers for all tables
op.execute(sa.text(create_refresh_updated_at_func.format(schema="foo")))
for table in MY_LIST_OF_TABLES:
op.execute(sa.text(create_trigger.format(schema="foo", table=table)))
def downgrade():
op.drop_table(..., schema="foo")
...
op.execute(sa.text("DROP FUNCTION foo.refresh_updated_at() CASCADE"))页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/70946151
复制相关文章
相似问题
腾讯云开发者
