55 lines
2.2 KiB
Python
55 lines
2.2 KiB
Python
# database.py
|
|
# 本文件负责配置数据库连接、创建引擎、会话工厂，并提供依赖注入函数
|
|
|
|
import os
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
# 1. 配置 MySQL 数据库连接 URL
|
|
# 格式：mysql+pymysql://用户名:密码@主机:端口/数据库名
|
|
# 优先读取环境变量（Docker / 生产环境注入），未设置时回退到本地默认值，方便本地开发。
|
|
# 例：Docker 中 DB_HOST 应设为 compose 里的服务名 "db"，而不是 localhost。
|
|
DB_USER = os.getenv("DB_USER", "root")
|
|
DB_PASSWORD = os.getenv("DB_PASSWORD", "123456")
|
|
DB_HOST = os.getenv("DB_HOST", "localhost")
|
|
DB_PORT = os.getenv("DB_PORT", "3306")
|
|
DB_NAME = os.getenv("DB_NAME", "claude_sms_test")
|
|
|
|
# 也可以直接用一个完整 URL 覆盖上面的拼接结果：DATABASE_URL=mysql+pymysql://用户:密码@主机:端口/库名
|
|
SQLALCHEMY_DATABASE_URL = os.getenv(
|
|
"DATABASE_URL",
|
|
f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}",
|
|
)
|
|
|
|
# 2. 创建数据库引擎
|
|
# - pool_pre_ping=True 表示每次从连接池取出连接前先 ping 一下，防止使用已断开的连接
|
|
# - pool_recycle=3600 表示连接空闲超过 1 小时自动回收，避免被 MySQL 主动断开
|
|
engine = create_engine(
|
|
SQLALCHEMY_DATABASE_URL,
|
|
pool_pre_ping=True,
|
|
pool_recycle=3600,
|
|
echo=os.getenv("SQL_ECHO", "false").lower() == "true" # 生产环境默认关闭 SQL 日志
|
|
)
|
|
|
|
# 3. 创建会话工厂
|
|
# autocommit=False：不自动提交，需要手动 commit
|
|
# autoflush=False：不自动 flush
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
# 4. 创建声明式基类，所有模型类都继承自它
|
|
Base = declarative_base()
|
|
|
|
# 5. 依赖注入函数：用于 FastAPI 路由中获取数据库会话
|
|
def get_db():
|
|
"""
|
|
每次请求创建一个数据库会话，请求结束后关闭。
|
|
这个函数会作为 Depends 的参数注入到路由中。
|
|
"""
|
|
db = SessionLocal() # 创建一个会话实例
|
|
try:
|
|
yield db # 将会话交给路由函数使用
|
|
finally:
|
|
db.close() # 无论是否发生异常，最后都会关闭会话，释放连接
|