FastAPI+SQLAlchemy操作SQLite时并发写入避免“数据库已锁定”错误的方案咨询

问题分析 #

首先得拆解清楚锁错误的根源：
SQLite是单写型数据库，同一时间仅允许一个写入事务持有锁。你的测试代码里，session_b通过db.flush()提前触发了SQL执行（数据写入数据库但未提交），直接抢占了SQLite的写入锁；而session_a等待0.5秒后执行commit时，发现锁已被session_b持有，再加上SQLAlchemy默认的SQLite连接锁等待逻辑很激进（几乎不等待就报错），所以直接抛出了锁定异常。

另外你配置了connect_args={"check_same_thread": False}，这打破了SQLite原生的线程安全限制，再结合FastAPI异步接口的事件循环线程调度机制，多个请求复用线程的情况进一步提升了锁冲突的概率。

安全可靠的解决方案 #

下面给出几个从易到难、适配不同场景的方案：

方案1：给SQLite连接添加锁超时配置

SQLite原生支持设置锁等待超时，我们可以在创建engine时加入timeout参数，让连接在获取锁失败时等待一段时间，而非直接报错。等持有锁的事务提交后，等待的事务就能自动获取锁完成操作。

修改engine创建代码：

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False, "timeout": 30}  # 最多等待30秒再报错
)

优点：改动极小，无业务代码侵入，对现有逻辑零影响。
缺点：高并发场景下会拉长接口响应时间，更适合低并发的内部工具或原型项目。

方案2：给commit操作添加重试机制

针对commit时可能出现的锁错误，我们可以封装一个重试逻辑，当捕获到OperationalError且错误信息包含锁提示时，自动重试几次。

先写一个简易的重试工具函数：

from sqlalchemy.exc import OperationalError
import asyncio

async def retry_commit(db, max_retries=3, delay=0.2):
    for _ in range(max_retries):
        try:
            db.commit()
            return True
        except OperationalError as e:
            if "database is locked" in str(e):
                await asyncio.sleep(delay)
                continue
            raise
    return False

再在接口中调用这个函数：

@app.post("/session_a")
async def session_a(db: Session = Depends(get_db)):
    logger.info("A start")
    uuid_str = str(uuid.uuid4())
    item = Item(name=f"session_a{uuid_str}")
    db.add(item)
    await asyncio.sleep(0.5)
    logger.info(f"A commit {uuid_str}")
    
    await retry_commit(db)
    return {"status": "A committed"}

优点：针对性强，仅处理锁冲突场景，不干扰正常业务流程。
缺点：需要修改业务代码，高并发下仍存在等待，但比方案1更灵活。

方案3：使用单线程连接池控制并发

可以改用SQLAlchemy的SingletonThreadPool连接池，确保同一时间只有一个连接操作数据库，从根源上避免锁竞争。同时如果你担心autoflush=False影响其他接口，可以只在长事务接口中临时禁用自动刷新。

修改连接池配置：

from sqlalchemy.pool import SingletonThreadPool

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},
    poolclass=SingletonThreadPool  # 单线程连接池，同一时间仅一个连接可用
)
SessionLocal = sessionmaker(autocommit=False, autoflush=True, bind=engine)

优点：从连接层面控制并发，彻底消除锁冲突，适合中小并发场景。
缺点：会牺牲部分并发性能，同一时间仅能处理一个写入请求。

方案4：将写入操作移至线程池执行

FastAPI的异步接口中，若存在阻塞IO操作（如数据库commit），可以将其放入asyncio.to_thread中执行，结合重试机制，既不阻塞事件循环，也能更好地处理锁等待。

修改接口代码：

@app.post("/session_a")
async def session_a(db: Session = Depends(get_db)):
    logger.info("A start")
    uuid_str = str(uuid.uuid4())
    item = Item(name=f"session_a{uuid_str}")
    db.add(item)
    await asyncio.sleep(0.5)
    logger.info(f"A commit {uuid_str}")
    
    # 把commit移到线程池执行，避免阻塞事件循环
    def commit_task():
        for _ in range(3):
            try:
                db.commit()
                return
            except OperationalError as e:
                if "database is locked" in str(e):
                    time.sleep(0.2)
                    continue
                raise
    await asyncio.to_thread(commit_task)
    return {"status": "A committed"}

优点：不阻塞FastAPI事件循环，同时兼容锁冲突处理，适合异步架构下的中高并发场景。
缺点：需要修改业务代码，对开发者的异步编程逻辑有一定要求。

方案5：升级为多写型数据库（终极方案）

如果你的项目后续有高并发写入需求，SQLite的单写特性会成为天然瓶颈，此时最好的方案是替换为PostgreSQL、MySQL这类支持多写的数据库，从根源上解决SQLite的锁限制。

优点：彻底解决并发写入问题，适配生产环境高并发场景。
缺点：需要额外部署和维护数据库服务，增加运维成本。

总结 #

• 低并发内部工具/原型项目：优先选方案1或方案3，改动小成本低；
• 异步架构中并发场景：方案2+方案4的组合是安全且灵活的选择；
• 生产环境高并发需求：直接升级到多写型数据库（方案5）是长远之计。

内容来源于stack exchange