dingdanquanliucheng/backend/app/db.py

"""数据库连接与会话管理模块。

提供 SQLAlchemy 引擎、会话工厂和依赖注入入口。
所有数据库操作通过 get_db_session() 获取会话。
"""

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

from backend.app.core.config import get_settings

settings = get_settings()


class Base(DeclarativeBase):
    """所有 ORM 模型的基类，声明式映射的基础。"""
    pass


def build_database_url() -> str:
    """根据配置项拼接 MySQL 连接 URL。

    Returns:
        str: pymysql 驱动格式的数据库连接字符串。
    """
    return (
        f"mysql+pymysql://{settings.mysql_user}:{settings.mysql_password}"
        f"@{settings.mysql_host}:{settings.mysql_port}/{settings.mysql_database}"
    )


engine = create_engine(build_database_url(), echo=settings.sql_echo, future=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)


def get_db_session():
    """提供数据库会话依赖，路由和 service 可复用这一入口。

    作为 FastAPI Depends 注入使用，请求结束后自动关闭会话。

    Yields:
        Session: SQLAlchemy 数据库会话实例。
    """
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()