Мне нужен совет о том, надежна ли моя текущая конфигурация и шаблон использования базы данных, и может ли она создавать узкие места, связанные с обработкой соединений, производительностью или журналированием в качестве приложения масштабируется.
Текущая настройка (высокий уровень)
- Бэкенд FastAPI
- Реляционная база данных (например, PostgreSQL/MySQL)
- SQLAlchemy для доступа к ORM/DB
- Взаимодействие с БД на основе сеанса по запросу
- Включено ведение журнала для запросов и событий приложения
Точка входа
Код: Выделить всё
""" Application entry point. """
from __future__ import annotations
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.config import settings
from app.database import check_database_health, dispose_engine, get_engine
from app.api.v1 import api_router
# Configure module logger
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
logger.info("Starting %s in %s mode...", settings.app_name, settings.environment)
# --- STARTUP PHASE ---
try:
if not check_database_health():
if settings.environment == "production":
logger.critical("Startup Failed: Database unreachable.")
raise RuntimeError("Database connection required in production mode.")
else:
logger.warning("Database unavailable. Running in limited mode.")
else:
# Attach engine to app state for access in request handlers
app.state.db_engine = get_engine()
logger.info("Database connection established: %s", settings.db_host or "from DATABASE_URL")
logger.info("Application Startup Complete.")
except Exception as e:
logger.critical("Startup Failed: %s", e)
raise
# --- RUNNING PHASE ---
yield
# --- SHUTDOWN PHASE ---
logger.info("Shutting down application...")
dispose_engine()
logger.info("Goodbye!")
def create_application() -> FastAPI:
app = FastAPI(
title=settings.app_name,
version=settings.app_version,
description="Production-grade derivatives pricing calculation engine",
lifespan=lifespan,
docs_url=f"{settings.api_base_path}/docs",
redoc_url=f"{settings.api_base_path}/redoc",
openapi_url=f"{settings.api_base_path}/openapi.json",
)
# --- CORS Middleware ---
if settings.allowed_origins:
app.add_middleware(
CORSMiddleware,
allow_origins=settings.allowed_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
logger.info("CORS enabled for %d origins.", len(settings.allowed_origins))
# --- Router Registration ---
app.include_router(
api_router,
prefix=settings.api_base_path,
)
logger.info("API routes registered under: %s", settings.api_base_path)
# --- Health Check Endpoint (outside API prefix for k8s probes) ---
@app.get("/health", tags=["Health"])
def health_check():
db_healthy = check_database_health()
return {
"status": "healthy" if db_healthy else "degraded",
"database": "connected" if db_healthy else "disconnected",
"version": settings.app_version,
}
@app.get(f"{settings.context_path}/health", tags=["Health"])
def health_check_with_context():
db_healthy = check_database_health()
return {
"status": "healthy" if db_healthy else "degraded",
"database": "connected" if db_healthy else "disconnected",
"version": settings.app_version,
}
return app
# Create the application instance
app = create_application()
Код: Выделить всё
from __future__ import annotations
import logging
from contextlib import contextmanager
from typing import Generator
from sqlalchemy import create_engine, text
from sqlalchemy.engine import Engine
from sqlalchemy.exc import OperationalError, SQLAlchemyError
from sqlalchemy.orm import sessionmaker, Session
from app.config import settings
logger = logging.getLogger(__name__)
def create_db_engine() -> Engine:
db_uri = settings.sqlalchemy_database_uri
if not db_uri:
logger.warning("No database URI configured. Engine creation skipped.")
return None
logger.info("Creating database engine for host: %s", settings.db_host or "from DATABASE_URL")
engine = create_engine(
db_uri,
pool_size=5,
max_overflow=10,
pool_pre_ping=True,
pool_recycle=1800, # 30 minutes
echo=settings.debug, # SQL logging only in debug mod
)
logger.info("Database engine created successfully.")
return engine
# Module-level engine instance (lazy initialization)
_engine: Engine | None = None
def get_engine() -> Engine:
global _engine
if _engine is None:
_engine = create_db_engine()
return _engine
def get_session_factory() -> sessionmaker:
engine = get_engine()
if engine is None:
raise RuntimeError("Database engine not initialized. Check database configuration.")
return sessionmaker(autocommit=False, autoflush=False, bind=engine)
@contextmanager
def get_db_session() -> Generator[Session, None, None]:
SessionLocal = get_session_factory()
session = SessionLocal()
try:
yield session
session.commit()
except SQLAlchemyError as e:
session.rollback()
logger.error("Database session error: %s", e)
raise
finally:
session.close()
def get_db() -> Generator[Session, None, None]:
SessionLocal = get_session_factory()
db = SessionLocal()
try:
yield db
except Exception:
db.rollback()
raise
finally:
db.close()
def check_database_health() -> bool:
engine = get_engine()
if engine is None:
logger.warning("Health check skipped: No database engine configured.")
return False
try:
with engine.connect() as connection:
connection.execute(text("SELECT 1"))
logger.info("Database health check passed.")
return True
except OperationalError as e:
logger.error("Database health check failed (OperationalError): %s", e)
return False
except Exception as e:
logger.error("Database health check failed (unexpected): %s", e)
return False
def dispose_engine() -> None:
global _engine
if _engine is not None:
try:
_engine.dispose()
logger.info("Database engine disposed successfully.")
except Exception as e:
logger.error("Error disposing database engine: %s", e)
finally:
_engine = None
Код: Выделить всё
import logging
import json # Import at top level
from typing import List, Union
from urllib.parse import quote_plus
from pydantic import Field, SecretStr, field_validator, computed_field, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
# 1. Configure Logging so INFO messages actually show up
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
class Settings(BaseSettings):
# Application Settings
app_name: str = "Pricing Calculation Engine"
app_version: str = "1.0.0"
environment: str = Field(default="production")
debug: bool = Field(default=False)
# Routing
context_path: str = "/calculation-engine"
api_prefix: str = "/api/v1"
@computed_field
@property
def api_base_path(self) -> str:
return f"{self.context_path}{self.api_prefix}"
# Database Creds
database_url: SecretStr = SecretStr("")
db_host: str | None = None
db_port: int = 5432
db_name: str | None = None
db_username: str | None = None
db_password: SecretStr | None = None
@computed_field
@property
def sqlalchemy_database_uri(self) -> str:
if self.database_url.get_secret_value():
return self.database_url.get_secret_value()
if not all([self.db_host, self.db_name, self.db_username, self.db_password]):
if self.environment == "production":
raise ValueError("Missing Database Config: Set DATABASE_URL or component fields.")
return ""
user = quote_plus(self.db_username)
password = quote_plus(self.db_password.get_secret_value())
return f"postgresql://{user}:{password}@{self.db_host}:{self.db_port}/{self.db_name}"
# Security
secret_key: SecretStr = SecretStr("")
algorithm: str = "HS256"
access_token_expire_minutes: int = 30
# CORS
allowed_origins: List[str] = Field(default_factory=list)
@field_validator("allowed_origins", mode="before")
@classmethod
def parse_cors_origins(cls, v: Union[str, List[str], None]) -> List[str]:
if v is None:
return []
if isinstance(v, list):
return v
if isinstance(v, str):
v = v.strip()
if v.startswith("[") and v.endswith("]"):
try:
return json.loads(v)
except json.JSONDecodeError:
pass
#Fallback to comma separation
return [origin.strip() for origin in v.split(",") if origin.strip()]
return []
# Validation
@model_validator(mode='after')
def verify_configuration(self) -> 'Settings':
if self.environment == "production" and not self.secret_key.get_secret_value():
raise ValueError("CRITICAL: 'SECRET_KEY' is missing.")
if self.environment == "production" and not self.database_url.get_secret_value():
if not all([self.db_host, self.db_name, self.db_username, self.db_password]):
raise ValueError("Database configuration incomplete.")
if self.environment == "production" and not self.allowed_origins:
logging.warning("No CORS origins configured for production.")
return self
model_config = SettingsConfigDict(
env_file=".env",
case_sensitive=False,
extra="ignore"
)
# Execution Block
try:
settings = Settings()
logging.info(f"Server starting in {settings.environment} mode.")
logging.info(f"CORS Origins loaded: {len(settings.allowed_origins)}")
except ValueError as e:
logging.error(f"Configuration Error: {e}")
exit(1)
except Exception as e:
logging.error(f"Unexpected Error during startup: {e}")
exit(1)
Учитывая мою текущую конфигурацию:
- Является ли эта база данных готовой к использованию и масштабируемой?
- Есть ли анти-шаблоны или риски, которые мне следует устранить сейчас?
- Какова будет рекомендация по доступу к базе данных через FastAPI для частых операций CRUD?
Подробнее здесь: https://stackoverflow.com/questions/798 ... lchemy-2-0
