SQLAlchemy+PostgreSQL单元测试间歇报错：删库后建库提示库已存在

开发者阿泽

2026-8-30

问题根因

这个偶发冲突不是DROP DATABASE逻辑本身失效，是三个典型问题叠加导致的，在CI流水线这种高并发、资源调度不稳定的环境下特别容易触发：

DROP DATABASE IF EXISTS 仅会屏蔽「目标库不存在」的报错，完全不会处理有活跃连接、拿不到库级排他锁的异常。你写的drop_database_connections一般是调用pg_terminate_backend踢除现有连接，但这个函数发起终止请求就会直接返回，不会等所有连接完全断开。你踢完连接立刻执行DROP时，只要还有1个没释放的连接占着目标库，DROP就会直接执行失败，库根本没被删掉，你没做异常捕获就直接往下走CREATE逻辑，自然触发唯一键冲突。
SQLAlchemy 1.4版本默认的连接上下文不会自动提交，而DROP DATABASE/CREATE DATABASE属于实例级DDL，不能在普通事务块内执行。你在with engine.connect()上下文里直接执行语句时，默认会开启一个事务，前面的踢连接、删库操作都在事务块内，没有手动执行commit()的话，部分场景下语句不会实际生效，就会出现你以为删了库、实际库还在的情况。
如果你开了pytest并行执行（比如用了pytest-xdist），多个worker进程会同时跑这套fixture逻辑，同时操作同名的test库，必然会出现时序冲突：A进程刚删完库还没来得及建，B进程又发起删/建请求，或者两个进程同时执行CREATE，就会随机出现报错。

修复方案

按优先级调整代码即可彻底解决问题：

首先把执行DDL的连接切换到自动提交隔离级别，库级操作不能放在普通事务里跑，同时给删库、建库逻辑加短重试，等连接完全释放、锁释放后再执行下一步：

import time
from sqlalchemy.exc import IntegrityError

@pytest.fixture
def apply_migrations():
    # 开自动提交，适配DDL执行要求
    with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
        # 多踢几次连接，等残留连接完全断开
        for _ in range(5):
            drop_database_connections(conn, cfg.POSTGRES_DB)
            time.sleep(0.2)
        
        # 删库加重试，处理偶发的锁占用问题
        for _ in range(10):
            try:
                conn.execute(f'DROP DATABASE IF EXISTS "{cfg.POSTGRES_DB}"')
                break
            except Exception as e:
                # 只捕获有其他连接占用的异常，其余异常直接抛出
                if "being accessed by other users" in str(e):
                    time.sleep(0.3)
                    continue
                raise
        
        # 建库加兜底重试
        for _ in range(10):
            try:
                conn.execute(f'CREATE DATABASE "{cfg.POSTGRES_DB}" TEMPLATE {template_db_name}')
                break
            except IntegrityError as e:
                if "pg_database_datname_index" in str(e):
                    # 碰到库已存在就再删一次重试
                    conn.execute(f'DROP DATABASE IF EXISTS "{cfg.POSTGRES_DB}"')
                    time.sleep(0.3)
                    continue
                raise

如果用了pytest-xdist跑并行测试，不要让所有worker共用一个固定的test库名，给每个worker分配独立的库名，从根源上避免多进程操作冲突：

import os

# 获取当前pytest worker的唯一标识
def get_worker_id() -> str:
    return os.environ.get("PYTEST_XDIST_WORKER", "main")

@pytest.fixture(scope="session")
def test_db_name(get_worker_id):
    return f"{cfg.POSTGRES_DB}_{get_worker_id}"

额外注意：用模板库建库时，要保证模板库没有任何活跃连接，不然CREATE DATABASE拿不到模板库的共享锁也会偶发失败，同样可以用上面的短重试逻辑兜底。

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