fix: P3-1 unify database config with app.config
Deploy / Deploy Staging (push) Failing after 1s
Deploy / Build Production Runtime Images (push) Has been skipped
Deploy / Deploy Production (push) Has been skipped
Deploy / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Validate Code Quality And Tests (push) Has been cancelled
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 1m0s
Tests / test (pull_request) Failing after 1m0s
Tests / lint (pull_request) Failing after 1m0s

This commit is contained in:
2026-06-26 17:56:33 +08:00
parent 4fefed332a
commit 3a97ca385a
+41 -26
View File
@@ -1,34 +1,49 @@
import os
from typing import Optional
"""Database session management and engine configuration.
from pydantic_settings import BaseSettings, SettingsConfigDict
统一使用 app.config 中的数据库配置,移除重复的 DatabaseSettings。
"""
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from contextlib import contextmanager
from typing import Generator
from app.config import settings
class DatabaseSettings(BaseSettings):
database_url: str = "postgresql+psycopg://postgres:postgres@postgres:5432/xiaoxia_saas"
pool_size: int = 20
max_overflow: int = 40
pool_timeout: int = 30
pool_recycle: int = 3600
engine = create_engine(
settings.database_url,
pool_size=settings.DATABASE_POOL_SIZE,
max_overflow=settings.DATABASE_MAX_OVERFLOW,
pool_timeout=settings.DATABASE_POOL_TIMEOUT,
pool_recycle=settings.DATABASE_POOL_RECYCLE,
)
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=False,
extra="ignore",
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
_settings: Optional[DatabaseSettings] = None
def get_db() -> Generator[Session, None, None]:
"""Dependency for getting database sessions."""
db = SessionLocal()
try:
yield db
finally:
db.close()
def get_database_settings() -> DatabaseSettings:
global _settings
if _settings is None:
env = os.getenv("APP_ENV", "development")
env_file = f".env.{env}" if env != "development" else ".env"
if os.path.exists(env_file):
_settings = DatabaseSettings(_env_file=env_file)
else:
_settings = DatabaseSettings()
return _settings
@contextmanager
def get_db_context() -> Generator[Session, None, None]:
"""Context manager for database sessions.
Usage:
with get_db_context() as db:
db.query(Model).all()
"""
db = SessionLocal()
try:
yield db
db.commit()
except Exception:
db.rollback()
raise
finally:
db.close()