wish/FastApi/app/core/db.py

"""SQLAlchemy database setup for MySQL.

Reads connection info from environment variables to avoid committing secrets.
- DATABASE_URL: full SQLAlchemy URL (optional)
- DB_USER, DB_PASS, DB_HOST, DB_PORT, DB_NAME: used when DATABASE_URL is not set

Example to set DATABASE_URL locally (Windows PowerShell):
$env:DATABASE_URL = "mysql+pymysql://root:%40%40qinys12346..@192.168.101.7:3306/wishing_tree?charset=utf8mb4&serverTimezone=Asia/Shanghai&use_unicode=true&characterEncoding=utf-8"

Install dependencies: pip install sqlalchemy pymysql
"""
import os
from typing import Generator

from sqlalchemy import create_engine
from sqlalchemy.engine import URL
from sqlalchemy.orm import sessionmaker, declarative_base

# Read connection info from environment variables
DATABASE_URL = os.getenv("DATABASE_URL")
DB_USER = os.getenv("DB_USER", "root")
DB_PASS = os.getenv("DB_PASS")
DB_HOST = os.getenv("DB_HOST", "192.168.1.177")
DB_PORT = int(os.getenv("DB_PORT", "3306"))
DB_NAME = os.getenv("DB_NAME", "wishing_tree")

# Query params for pymysql / SQLAlchemy. Keep only supported options (charset).
QUERY = {
    "charset": "utf8mb4",
}

if DATABASE_URL:
    # If the user set a full SQLAlchemy URL, prefer it
    engine_url = DATABASE_URL
else:
    # Build a SQLAlchemy URL object (handles proper quoting)
    engine_url = URL.create(
        drivername="mysql+pymysql",
        username=DB_USER,
        password=DB_PASS,
        host=DB_HOST,
        port=DB_PORT,
        database=DB_NAME,
        query=QUERY,
    )

# Create engine and session factory
# Tune pool parameters as needed for your environment
engine = create_engine(
    engine_url,
    pool_pre_ping=True,
    pool_size=10,
    max_overflow=20,
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


# Dependency for FastAPI endpoints
def get_db() -> Generator:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()