Flask对接Postgresql出现Table (view) already exists报错如何优雅解决

运行环境 #

• Flask version : 2.0.1
• Postgresql: 13.3

现有代码说明 #

视图定义代码 #

class MarketView(db.Model):
    __table__ = create_view(
        name="market_view",
        selectable=select(
            [
                Tokens.id,
                Tokens.symbol,
                Tokens.name,
                Tokens.tags,
                Tokens.max_supply,
                Tokens.circulating_supply,
                Tokens.total_supply,
                Tokens.ecosystem,
                Tokens.consensus,
                Tokens.category,
                Tokens.portfolio,
                Market.usd,
                Market.btc,
                Market.cmc_rank,
                Market.trust_score,
                Indicator.rsi,
                Indicator.stochastic,
                Indicator.bollinger,
                Indicator.cci,
                Indicator.cfi,
                Indicator.ema,
                Indicator.evm,
                Indicator.macd,
                Indicator.roc,
            ],
            from_obj=(
                Tokens.__table__.join(
                    Market, Tokens.id == Market.token_id
                ).join(Indicator, Tokens.id == Indicator.token_id)
            ),
        ),
        metadata=db.metadata,
    )

数据库初始化代码 #

def configure_database(app):
    @app.before_first_request
    def initialize_database():
        try:
            db.create_all()
        except ProgrammingError:
            pass

    @app.teardown_request
    def shutdown_session(exception=None):
        db.session.remove()

优雅解决方案 #

方案1：修改create_view函数，新增不存在才创建的逻辑 #

你使用的create_view函数默认生成的是CREATE VIEW语句，只要修改为适配PostgreSQL的语法，增加不存在判断即可，改造成本最低：

def create_view(name, selectable, metadata):
    # 原有逻辑保留，仅修改DDL语句部分
    # 不需要覆盖旧视图定义用这个：
    create_view_ddl = f"CREATE VIEW IF NOT EXISTS {name} AS {selectable}"
    # 如果视图逻辑会迭代更新，需要覆盖旧定义用这个：
    # create_view_ddl = f"CREATE OR REPLACE VIEW {name} AS {selectable}"
    db.DDL(create_view_ddl).execute_at('after_create', metadata)
    # 后续返回表对象的原有逻辑保留

注意：CREATE OR REPLACE VIEW要求新的视图返回的字段名、字段类型和旧定义完全兼容，否则会执行失败

方案2：创建前先检查视图是否存在，过滤已存在对象 #

在调用db.create_all()之前先查询PG系统表判断视图是否存在，已存在就临时从metadata中移除，避免重复创建：

@app.before_first_request
def initialize_database():
    # 检查public schema下是否存在目标视图
    view_exists = db.session.query(
        db.text("exists(select 1 from pg_views where schemaname = 'public' and viewname = 'market_view')")
    ).scalar()
    if view_exists:
        # 临时移除视图，避免create_all尝试重复创建
        db.metadata.remove(MarketView.__table__)
    db.create_all()
    # 若后续需要用到完整metadata结构，可以再加回来
    if view_exists:
        db.metadata._add_table(MarketView.__table__.name, MarketView.__table__.schema, MarketView.__table__)

方案3：使用Flask-Migrate管理数据库变更（生产环境标准方案） #

如果项目需要长期迭代数据库结构，直接用Flask-Migrate（基于Alembic）管理所有表和视图的创建、变更逻辑，完全替换db.create_all()的自动创建逻辑，从根源上避免重复创建问题，同时支持结构变更回滚、多环境结构同步。
操作逻辑：

1. 安装Flask-Migrate后初始化迁移目录
2. 生成初始迁移文件，手动添加视图的创建逻辑
3. 后续每次表或视图结构变更都生成新的迁移文件，上线时执行迁移即可

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