在SQLModel应用中，所有数据库操作都依赖一个核心对象：Engine。这个对象负责维护与数据库的连接池，管理事务和执行的上下文。如果在每次业务请求或每次实例化DB类时都重新创建Engine，连接池会被反复建立和销毁，带来不必要的握手开销和资源浪费。更严重的是，当并发量上升时，每个独立引擎都会持有自己的连接池，数据库连接数可能迅速攀升至上限，导致服务不可用。因此，如何在DB包装类中有效共享引擎成为架构设计的关键一环。
共享引擎的核心矛盾与常见误用
SQLModel基于SQLAlchemy Core构建，create_engine函数返回的Engine实例本身就是线程安全的，内部封装了连接池。官方文档推荐在应用生命周期内只创建一次引擎，后续所有会话都从该引擎获取连接。然而，很多开发者习惯将引擎创建放在DB类的__init__方法中，例如：
class Database:
def __init__(self, url: str):
self.engine = create_engine(url)
# 每次实例化都会创建一个新引擎
这种写法在小流量或脚本任务中可能看不出问题，但一旦部署到Web服务，每个请求处理函数如果都实例化Database，就会创建大量引擎。即使使用单例模式限制了实例数量，如果多个DB类分别创建引擎，仍然无法共享连接池。以FastAPI为例，依赖注入默认会为每个请求调用一次可调用对象，如果直接在依赖函数中create_engine，同样会重复创建。
另一个常见误区是使用全局变量但忽视了模块导入顺序和测试环境。全局引擎在测试时可能指向生产数据库，或者被多个测试用例共享导致数据污染。因此，共享策略不仅要考虑运行时效率，还要兼顾可测试性和环境隔离。
三种实用的引擎共享策略
针对不同架构，可以选择以下几种经过验证的共享方式。第一种是模块级单例，即在独立模块中创建引擎并暴露一个获取函数，所有DB类都导入该模块并使用同一个引擎。
# db/engine.py
from sqlmodel import create_engine
_engine = None
def get_engine():
global _engine
if _engine is None:
DATABASE_URL = "postgresql://user:pass@localhost/db"
_engine = create_engine(DATABASE_URL, pool_size=5, max_overflow=10)
return _engine
这种方式简单直接，但需要确保在应用启动时初始化，避免并发首次调用时产生竞争条件。可以通过加锁或使用lru_cache等方式改进。
第二种是类属性缓存，将引擎作为类变量存储，所有实例共享。这种方式适合DB类被多次实例化但需要共享引擎的场景。
from sqlmodel import create_engine, Session
class Database:
_engine = None
def __init__(self, url: str):
if Database._engine is None:
Database._engine = create_engine(url, pool_pre_ping=True)
@property
def engine(self):
return Database._engine
def get_session(self):
return Session(self.engine)
类属性缓存保证了无论创建多少个Database对象，底层引擎只有一个。但要注意，如果传入不同的数据库URL，第二次实例化不会生效，因为引擎已经固定。因此这种方式适合单一数据库连接的应用。
第三种是依赖注入模式，尤其适合FastAPI等框架。将引擎创建放在应用启动事件中，存入app.state，然后在依赖函数中读取。
from fastapi import FastAPI, Depends
from sqlmodel import create_engine, Session
app = FastAPI()
@app.on_event("startup")
def init_engine():
app.state.engine = create_engine("sqlite:///app.db", pool_pre_ping=True)
def get_session():
with Session(app.state.engine) as session:
yield session
@app.get("/items")
def read_items(session: Session = Depends(get_session)):
# 使用session查询
pass
依赖注入让引擎的生命周期与框架生命周期绑定，便于管理，也容易在测试时替换为不同的引擎。但需要注意异步应用与同步引擎的协调问题。
连接池调优与常见陷阱
共享引擎后，连接池的参数配置直接影响系统吞吐量。SQLAlchemy默认使用QueuePool，可以通过pool_size设置池中保持的连接数，max_overflow设置超出pool_size后最多可创建的连接数。如果设置过小，高并发下会出现等待超时；设置过大，则可能超出数据库的最大连接限制。
engine = create_engine(
"postgresql://user:pass@localhost/db",
pool_size=10, # 常驻连接数
max_overflow=20, # 额外可创建的连接数
pool_timeout=30, # 获取连接的超时时间（秒）
pool_recycle=1800, # 连接回收时间，防止数据库断开
pool_pre_ping=True # 每次取出连接前检查有效性
)
在多线程环境下，Engine和连接池是线程安全的，但Session不是。每个线程应当使用独立的Session，并在使用后关闭。如果使用异步SQLModel（如搭配aiosqlite或asyncpg），需要创建AsyncEngine，此时共享策略类似，但要注意异步引擎不能在多个事件循环之间共享。
另一个常见的坑是在测试中复用全局引擎导致数据相互影响。解决方案是为每个测试用例使用独立的临时数据库或回滚事务。例如使用SQLite内存数据库，每个测试创建新的引擎，或者通过嵌套事务实现回滚，保持引擎共享的同时隔离数据。
最后，当应用需要连接多个数据库时，应避免将所有引擎塞进一个大而全的全局字典，而是按需延迟创建并缓存，利用lru_cache或自定义缓存机制实现多引擎的共享管理。
SQLModel数据库引擎共享Python ORM修改时间：2026-09-17 19:36:01