使用 SQLAlchemy 缓存表状态答案

【问题标题】：Caching a table state using SQL Alchemy使用 SQLAlchemy 缓存表状态
【发布时间】：2022-01-27 19:54:11
【问题描述】：

我有一个不会经常更改的表，因此我希望将其缓存在客户端。那么是否有某种 hash value 或 last update time 可用于确定表是否已更新？如果没有，那么如何创建触发器来跟踪表中的更改？
Python - SQL Alchemy、Fast API、Postgres SQL

class Country(Base):
    __tablename__ = 'country'
    id:    int = Column(Integer,    primary_key=True)
class State(Base):
    __tablename__ = 'state'
    id:    int = Column(Integer,    primary_key=True)

class Cache(Base):
    __tablename__ = 'cache'
    index:          int      = Column(Integer, primary_key=True)
    table_name:     str      = Column(String(64), nullable=False, unique=True)
    sync_token:     str      = Column(String(40), nullable=False)

# triggers
@event.listens_for(Country, 'after_update')
def after_update(mapper, connection, target):
    pass '''on any change in this table update the timestamp in the cache table'''

【问题讨论】：

虽然你的情况是相当静态的 - 最好不要这样做。构建和维护/同步多组状态（与客户端会话一样多）既困难又昂贵。这是一条永远不会有回报的滑溜溜的路。遵守“单一事实来源”规则，只在数据库中保留一个状态。
@Stefanov.sm 表格不会有太大变化——就像一个国家的名称和州数不太可能改变一样。我想要这个，因为如果它在任何遥远的可能性中发生变化，那么客户端应该能够反映相同的变化。

标签： python database postgresql sqlalchemy fastapi

【解决方案1】：

您可以在表中添加任何修改时更新时间戳的列：

    from datetime import datetime
    from sqlalchemy import Column, DateTime, func

    modified_at = Column(
        DateTime,
        server_default=func.timezone("UTC", func.now()),
        onupdate=datetime.utcnow,
    )

您还可以使用事件来跟踪表格中的更改。这是一个涵盖它的主题：Tracking model changes in SQLAlchemy

from sqlalchemy.orm import sessionmaker

# triggers
@event.listens_for(Country, 'after_update')
def after_update(mapper, connection, target):
   # target is your updated Country instance
   connection.execute(f"update cache set index = {target.int} where table_name = 'country'")
   # last_update_on will be modified automatically

【讨论】：

我希望水平扩展这个解决方案。我认为触发器是最好的选择，因为它不依赖于 SQL Alchemy。我认为将 modified_at 列添加到我需要缓存的表中不会解决这个特定问题。如果您认为确实如此，请解释一下。
modified_at 是对此的回答：“那么是否有某种哈希值或上次更新时间可用于确定表是否已更新？”。您可以使用：在 db 层触发，在 sqlalchemy 或某种本地事件总线上的事件：github.com/Enforcer/pybuses。您需要某种 CQRS 解决方案。横向是什么意思？应用程序的水平节点会使用相同的数据库吗？
是的，由多个节点访问的单个数据库运行相同的 python 应用程序。是的，使用触发器或事件是完成此任务的正确方法，但不确定如何编写触发器或事件。你能用一个抽象的例子来修改你的答案吗？（我已经编辑了问题）谢谢。
我写了一个答案，但我不确定你的国家/州和缓存模型是否正确
sync_token 列将包含一个 SHA-1 字符串，该字符串会在每次更新具有 table_name 的表时发生变化。我已经设法让它发挥作用。谢谢

【解决方案2】：

class Cache(Base):
    __tablename__ = 'cache'
    index:          int = Column(Integer, primary_key=True)
    table_name:     str = Column(String(64), nullable=False, unique=True)   
    last_update_on: datetime = Column(DateTime, nullable=False, default=datetime.now)

# triggers
@event.listens_for(Country, 'after_update')
def run_after_update(mapper, connection, target):
    dateTime:  str = str(datetime.now())
    tableName: str = Country.__tablename__
    update_stmt = f"UPDATE cache SET last_update_on = '{dateTime}' WHERE table_name = '{tableName}'"
    connection.execute(update_stmt)

【讨论】：