使用Uvicorn运行SQLAlchemy：如何避免多Worker重复调用Metadata.create_all？

我正在开发基于FastAPI、SQLAlchemy、Pydantic/psycopg2和Uvicorn栈的Python应用，项目结构参考FastAPI官方「大型应用-多文件」示例。项目中创建了包含所有应用表的SQLAlchemy MetaData对象Base，以及SQLAlchemy Engine对象engine，并在主函数中调用Base.metadata.create_all(bind=engine)，通过uvicorn命令启动服务。

单ASGI Worker首次启动正常，但使用多Worker（如--workers 4）时，部分Worker会在首个Worker完成元数据创建前执行该调用，导致报错：

sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "pg_class_relname_nsp_index"
DETAIL:  Key (relname, relnamespace)=(accounts_id_seq, 2200) already exists.

[SQL: 
CREATE TABLE accounts (
    id SERIAL NOT NULL, 
    PRIMARY KEY (id)
)
]
(Background on this error at: https://sqlalche.me/e/20/gkpj)

这个错误虽在意料之中但不符合预期，目前我只能先用单Worker初始化空PostgreSQL数据库，这更像是临时workaround，想了解是否有更合适的解决方案？

1. 用数据库迁移工具提前初始化（推荐生产环境） #

把表创建从应用启动流程中完全剥离，使用SQLAlchemy官方的迁移工具Alembic来管理数据库结构：

• 初始化Alembic：alembic init alembic
• 修改alembic.ini中的数据库连接URL，和项目中的engine保持一致
• 生成初始迁移脚本：alembic revision --autogenerate -m "init tables"
• 执行迁移创建表：alembic upgrade head
• 之后启动Uvicorn多Worker时，不再需要在应用中调用Base.metadata.create_all，直接启动服务即可。

这种方式是生产环境的标准做法，不仅能避免多Worker冲突，还能管理后续的表结构变更，回滚也更安全。

2. 数据库级分布式锁保证单进程执行 #

如果不想剥离启动时的表初始化逻辑，可以用PostgreSQL的**咨询锁（Advisory Lock）**来确保同一时间只有一个Worker执行表创建操作：

from sqlalchemy import create_engine
from models import Base  # 你的Base对象

engine = create_engine("postgresql://user:password@host/db")

def init_db():
    with engine.connect() as conn:
        # 获取咨询锁，锁ID自定义一个唯一值即可
        conn.execute("SELECT pg_advisory_lock(123456)")
        try:
            Base.metadata.create_all(bind=conn)
        finally:
            # 释放锁
            conn.execute("SELECT pg_advisory_unlock(123456)")

# 在应用启动时调用init_db
init_db()

多个Worker启动时，只有第一个能拿到锁并执行表创建，其他Worker会等待锁释放，此时表已经存在，create_all会跳过创建逻辑，不会触发重复序列的错误。

3. 主进程提前初始化后启动Worker #

用Python代码而非命令行启动Uvicorn，让主进程先完成数据库初始化，再fork出Worker：

import uvicorn
from sqlalchemy import create_engine
from models import Base

# 主进程先初始化数据库
engine = create_engine("postgresql://user:password@host/db")
Base.metadata.create_all(bind=engine)

# 启动多Worker的Uvicorn服务
if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, workers=4)

Uvicorn的多Worker模式会基于主进程fork子进程，主进程已经完成了表创建，子进程启动时不需要再执行create_all，从根源避免了冲突。注意：如果项目中使用了SQLAlchemy的连接池，需要确保Worker进程启动时重新创建连接池（不过此方案中create_all只在主进程执行一次，Worker不需要再执行，所以不会有问题）。

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