如何通过SQLAlchemy在SQL数据库中创建自动更新的Union永久表

根据你的需求，要创建一个能自动同步equities和bonds表变更的联合表（仅包含id和currency），有两种主流实现方式，取决于你需要虚拟视图还是物理存储的物化视图：

方案1：数据库视图（Virtual View） #

视图是虚拟表，不会存储实际数据，查询时实时从源表获取数据，天然自动同步源表的所有变更，适合对实时性要求高、不需要频繁查询的场景。

SQLAlchemy 声明式实现 #

直接基于现有模型定义视图类，通过SQL语句创建视图：

import sqlalchemy as sa
import sqlalchemy.orm as orm

Base = orm.declarative_base()

# 原有的Equity和Bond类保持不变
class Equity(Base):
    __tablename__ = 'equities'
    
    id = sa.Column(sa.String, primary_key=True)
    name = sa.Column(sa.String, nullable=False)
    currency = sa.Column(sa.String, nullable=False)
    country = sa.Column(sa.String, nullable=False)
    sector = sa.Column(sa.String, nullable=True)
    
    def __repr__(self):
        return f"Equity('Ticker: {self.id}', 'Name: {self.name}')"

class Bond(Base):
    __tablename__ = 'bonds'
    
    id = sa.Column(sa.String, primary_key=True)
    name = sa.Column(sa.String, nullable=False)
    country = sa.Column(sa.String, nullable=False)
    currency = sa.Column(sa.String, nullable=False)
    sector = sa.Column(sa.String, nullable=False)
    
    def __repr__(self):
        return f"Bond('Ticker: {self.id}', 'Name: {self.name}')"

# 定义联合视图
class AssetCurrencyView(Base):
    __tablename__ = 'asset_currency_view'
    id = sa.Column(sa.String, primary_key=True)
    currency = sa.Column(sa.String, nullable=False)
    
    @classmethod
    def create_view(cls, engine):
        # 执行创建视图的SQL，UNION会自动去重，用UNION ALL可保留重复项
        create_view_sql = sa.text("""
            CREATE OR REPLACE VIEW asset_currency_view AS
            SELECT id, currency FROM equities
            UNION
            SELECT id, currency FROM bonds
        """)
        with engine.connect() as conn:
            conn.execute(create_view_sql)
            conn.commit()

# 初始化表和视图
if __name__ == "__main__":
    engine = sa.create_engine("postgresql://user:password@localhost/dbname")
    # 先创建源表
    Base.metadata.create_all(engine)
    # 再创建视图
    AssetCurrencyView.create_view(engine)

使用视图查询 #

和普通ORM模型一样操作：

session = orm.Session(engine)
assets = session.query(AssetCurrencyView.id, AssetCurrencyView.currency).all()

方案2：物化视图（Materialized View）+ 触发器（物理表实时同步） #

如果你需要一个物理存储的永久表（数据实际存在磁盘上，查询性能更高），同时要自动同步源表的变更，可以用物化视图配合触发器实现（以PostgreSQL为例，不同数据库语法略有差异）。

SQLAlchemy 实现代码 #

import sqlalchemy as sa
import sqlalchemy.orm as orm

Base = orm.declarative_base()

# 原有的Equity和Bond类保持不变
class Equity(Base):
    __tablename__ = 'equities'
    
    id = sa.Column(sa.String, primary_key=True)
    name = sa.Column(sa.String, nullable=False)
    currency = sa.Column(sa.String, nullable=False)
    country = sa.Column(sa.String, nullable=False)
    sector = sa.Column(sa.String, nullable=True)
    
    def __repr__(self):
        return f"Equity('Ticker: {self.id}', 'Name: {self.name}')"

class Bond(Base):
    __tablename__ = 'bonds'
    
    id = sa.Column(sa.String, primary_key=True)
    name = sa.Column(sa.String, nullable=False)
    country = sa.Column(sa.String, nullable=False)
    currency = sa.Column(sa.String, nullable=False)
    sector = sa.Column(sa.String, nullable=False)
    
    def __repr__(self):
        return f"Bond('Ticker: {self.id}', 'Name: {self.name}')"

# 定义物化视图对应的ORM类
class AssetCurrencyMaterializedView(Base):
    __tablename__ = 'asset_currency_mv'
    id = sa.Column(sa.String, primary_key=True)
    currency = sa.Column(sa.String, nullable=False)
    
    @classmethod
    def create_materialized_view(cls, engine):
        # 创建物化视图并初始化数据
        create_mv_sql = sa.text("""
            CREATE MATERIALIZED VIEW asset_currency_mv AS
            SELECT id, currency FROM equities
            UNION
            SELECT id, currency FROM bonds
            WITH DATA
        """)
        # 创建刷新物化视图的触发器函数
        create_trigger_func_sql = sa.text("""
            CREATE OR REPLACE FUNCTION refresh_asset_currency_mv()
            RETURNS TRIGGER AS $$
            BEGIN
                REFRESH MATERIALIZED VIEW asset_currency_mv;
                RETURN NULL;
            END;
            $$ LANGUAGE plpgsql;
        """)
        # 给equities表添加变更触发器
        create_equity_trigger_sql = sa.text("""
            CREATE TRIGGER trigger_equities_change
            AFTER INSERT OR UPDATE OR DELETE ON equities
            FOR EACH STATEMENT
            EXECUTE FUNCTION refresh_asset_currency_mv();
        """)
        # 给bonds表添加变更触发器
        create_bond_trigger_sql = sa.text("""
            CREATE TRIGGER trigger_bonds_change
            AFTER INSERT OR UPDATE OR DELETE ON bonds
            FOR EACH STATEMENT
            EXECUTE FUNCTION refresh_asset_currency_mv();
        """)
        
        with engine.connect() as conn:
            conn.execute(create_mv_sql)
            conn.execute(create_trigger_func_sql)
            conn.execute(create_equity_trigger_sql)
            conn.execute(create_bond_trigger_sql)
            conn.commit()

# 初始化表和物化视图
if __name__ == "__main__":
    engine = sa.create_engine("postgresql://user:password@localhost/dbname")
    Base.metadata.create_all(engine)
    AssetCurrencyMaterializedView.create_materialized_view(engine)

注意事项 #

• 物化视图存储实际数据，查询更快，但会占用额外存储资源
• 触发器会在源表变更时自动刷新物化视图，保证数据同步，但每次变更都会触发刷新，可能影响写入性能

内容的提问来源于stack exchange，提问作者Gcedism