dingdanquanliucheng/backend/app/db.py

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):
    pass


def build_database_url() -> str:
    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 可复用这一入口。"""
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()