推荐答案
在 FastAPI 中处理数据库连接通常使用依赖注入的方式，结合 SQLAlchemy 或 Databases 等库来管理数据库连接。以下是一个使用 SQLAlchemy 的示例：
// www.javascriptcn.com code example
from fastapi import FastAPI, Depends
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
# 数据库连接配置
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
# 创建数据库引擎
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
# 创建会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 声明基类
Base = declarative_base()
# 创建 FastAPI 应用
app = FastAPI()
# 依赖函数，用于获取数据库会话
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# 示例路由
@app.get("/items/{item_id}")
def read_item(item_id: int, db: Session = Depends(get_db)):
# 在这里可以使用 db 进行数据库操作
item = db.query(Item).filter(Item.id == item_id).first()
return item本题详细解读
1. 数据库连接配置
在 FastAPI 中，首先需要配置数据库连接。通常使用 SQLALCHEMY_DATABASE_URL 来指定数据库的 URL。例如，使用 SQLite 数据库时，URL 可以是 sqlite:///./test.db。
2. 创建数据库引擎
使用 create_engine 函数创建数据库引擎。这个引擎负责管理与数据库的连接。对于 SQLite，需要传递 connect_args={"check_same_thread": False} 以确保线程安全。
3. 创建会话工厂
通过 sessionmaker 创建一个会话工厂 SessionLocal。这个工厂用于生成数据库会话（Session），每个会话代表一个数据库连接。
4. 声明基类
使用 declarative_base 创建一个基类 Base，所有数据库模型都将继承自这个基类。
5. 创建 FastAPI 应用
使用 FastAPI() 创建一个 FastAPI 应用实例。
6. 依赖函数
定义一个依赖函数 get_db，用于获取数据库会话。这个函数使用 yield 来确保会话在使用后被正确关闭。
7. 示例路由
在路由中使用 Depends(get_db) 来注入数据库会话。这样，每个请求都会自动获取一个新的数据库会话，并在请求处理完成后关闭它。
通过这种方式，FastAPI 可以高效地管理数据库连接，确保每个请求都能获得独立的数据库会话，同时避免资源泄漏。