51 lines
2.1 KiB
Python
51 lines
2.1 KiB
Python
# ============================================================
|
||
# 数据库连接与会话管理
|
||
# 负责：
|
||
# 1. 构建 SQLAlchemy 引擎（engine），建立与 MySQL 的连接
|
||
# 2. 声明式基类 Base — 所有 ORM 模型都继承它
|
||
# 3. 提供 FastAPI 依赖注入用的 get_db() 会话生成器
|
||
# ============================================================
|
||
|
||
from sqlalchemy import *
|
||
from sqlalchemy.orm import declarative_base, sessionmaker
|
||
|
||
# MySQL 连接串格式: mysql+pymysql://用户名:密码@主机:端口/数据库?字符集
|
||
# 当前项目使用 pymysql 作为底层驱动，数据库为本地 student
|
||
db_url = "mysql+pymysql://root:123456@127.0.0.1:3306/student?charset=utf8mb4"
|
||
|
||
# 创建 SQLAlchemy 引擎：负责连接池管理、SQL 执行、事务提交等底层工作
|
||
# 所有 ORM 操作最终都经由 engine 下发到 MySQL
|
||
engine = create_engine(db_url)
|
||
|
||
# 声明式基类：ORM 模型必须继承它才能被 SQLAlchemy 识别为表
|
||
# main.py 中的 Base.metadata.create_all(engine) 会据此自动建表
|
||
Base = declarative_base()
|
||
|
||
# 会话工厂：每次调用 Session() 就会拿到一个全新的数据库会话
|
||
# autoflush=False —— 不自动 flush，由我们在事务边界手动控制写入
|
||
# autocommit=False —— 关闭自动提交，事务由调用方显式 commit / rollback
|
||
Session = sessionmaker(
|
||
bind=engine,
|
||
autoflush=False,
|
||
autocommit=False,
|
||
)
|
||
|
||
|
||
def get_db():
|
||
"""
|
||
FastAPI 依赖注入函数 —— 为每个请求生成一个独立的数据库会话。
|
||
|
||
使用 yield 构成生成器，前半段（yield 之前）在接口执行前拿到 Session，
|
||
后半段（yield 之后，finally 块）在接口执行完毕后自动 close，
|
||
确保无论接口正常返回还是抛异常，会话都不会泄漏。
|
||
|
||
典型用法（在路由函数里声明 db = Depends(get_db)）:
|
||
def some_api(db = Depends(get_db)):
|
||
... 使用 db.query(...) 查数据 ...
|
||
db.commit() # 有写入时手动提交
|
||
"""
|
||
db = Session()
|
||
try:
|
||
yield db
|
||
finally:
|
||
db.close() |