Caching a table state using SQL Alchemy-CodePudding

I have a table that won't change very frequently and hence I wish to cache it at the client side. So is there some sort of hash value or last update time which can be used to determine if the table is updated? If not, then how can I create a trigger to track the changes in tables?
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'''

CodePudding user response：

You can add to your table column that update timestamp when any modification in the table would have been made:

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

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

You can also use events to track changes in your table. Here is a topic covering it: 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

CodePudding user response：

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)