011-数据库连接与配置

难度：🟡 | 预计时间：90分钟 | 前置：010-WebSocket实时通信

学习目标

通过本章学习，您将能够：

• 掌握数据库集成：理解FastAPI与各种数据库的集成方式和最佳实践
• 配置连接管理：实现数据库连接池、会话管理和连接优化
• 设计数据模型：使用SQLAlchemy创建和管理数据库模型
• 实现CRUD操作：构建完整的数据库操作层和业务逻辑
• 优化数据库性能：掌握查询优化、索引设计和缓存策略
• 管理数据库迁移：实现数据库版本控制和自动化迁移

11.1 数据库选择与配置

11.1.1 支持的数据库类型

FastAPI通过SQLAlchemy支持多种数据库：

• 关系型数据库：PostgreSQL、MySQL、SQLite、Oracle、SQL Server
• NoSQL数据库：MongoDB（通过Motor）、Redis
• 内存数据库：SQLite（内存模式）

数据库类型对比

数据库类型	优势	劣势	适用场景	性能特点
PostgreSQL	ACID完整性、丰富特性、开源	配置复杂、资源消耗大	企业级应用、复杂查询	高并发、强一致性
MySQL	简单易用、生态丰富、高性能	功能相对简单	Web应用、中小型项目	读写分离、主从复制
SQLite	轻量级、无需服务器、嵌入式	并发限制、功能有限	开发测试、小型应用	低延迟、文件存储
MongoDB	灵活模式、水平扩展、文档存储	事务支持有限	内容管理、大数据	分片集群、高吞吐
Redis	内存存储、高性能、丰富数据结构	持久化限制、内存成本高	缓存、会话存储	极高性能、亚毫秒延迟

数据库连接生命周期

11.1.2 项目结构设计

app/
├── __init__.py
├── main.py
├── core/
│   ├── __init__.py
│   ├── config.py          # 配置管理
│   └── database.py        # 数据库配置
├── models/
│   ├── __init__.py
│   ├── base.py           # 基础模型
│   └── user.py           # 用户模型
├── schemas/
│   ├── __init__.py
│   └── user.py           # Pydantic模型
├── crud/
│   ├── __init__.py
│   ├── base.py           # 基础CRUD
│   └── user.py           # 用户CRUD
├── api/
│   ├── __init__.py
│   ├── deps.py           # 依赖项
│   └── v1/
│       ├── __init__.py
│       ├── api.py        # API路由汇总
│       └── endpoints/
│           ├── __init__.py
│           └── users.py  # 用户端点
└── db/
    ├── __init__.py
    ├── base.py           # 导入所有模型
    ├── init_db.py        # 数据库初始化
    └── session.py        # 会话管理

11.2 配置管理

11.2.1 环境配置

# app/core/config.py
from typing import Any, Dict, List, Optional, Union
from pydantic import AnyHttpUrl, BaseSettings, PostgresDsn, validator
import secrets

class Settings(BaseSettings):
    """应用配置"""
    
    # 基础配置
    API_V1_STR: str = "/api/v1"
    SECRET_KEY: str = secrets.token_urlsafe(32)
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8  # 8天
    SERVER_NAME: str = "FastAPI Server"
    SERVER_HOST: AnyHttpUrl = "http://localhost"
    
    # CORS配置
    BACKEND_CORS_ORIGINS: List[AnyHttpUrl] = []
    
    @validator("BACKEND_CORS_ORIGINS", pre=True)
    def assemble_cors_origins(cls, v: Union[str, List[str]]) -> Union[List[str], str]:
        if isinstance(v, str) and not v.startswith("["):
            return [i.strip() for i in v.split(",")]
        elif isinstance(v, (list, str)):
            return v
        raise ValueError(v)
    
    # 数据库配置
    POSTGRES_SERVER: str = "localhost"
    POSTGRES_USER: str = "postgres"
    POSTGRES_PASSWORD: str = "password"
    POSTGRES_DB: str = "fastapi_db"
    POSTGRES_PORT: str = "5432"
    
    SQLALCHEMY_DATABASE_URI: Optional[PostgresDsn] = None
    
    @validator("SQLALCHEMY_DATABASE_URI", pre=True)
    def assemble_db_connection(cls, v: Optional[str], values: Dict[str, Any]) -> Any:
        if isinstance(v, str):
            return v
        return PostgresDsn.build(
            scheme="postgresql",
            user=values.get("POSTGRES_USER"),
            password=values.get("POSTGRES_PASSWORD"),
            host=values.get("POSTGRES_SERVER"),
            port=values.get("POSTGRES_PORT"),
            path=f"/{values.get('POSTGRES_DB') or ''}",
        )
    
    # SQLite配置（开发环境）
    SQLITE_DATABASE_URI: str = "sqlite:///./app.db"
    
    # 数据库连接池配置
    DB_POOL_SIZE: int = 20
    DB_MAX_OVERFLOW: int = 30
    DB_POOL_TIMEOUT: int = 30
    DB_POOL_RECYCLE: int = 3600
    
    # Redis配置
    REDIS_HOST: str = "localhost"
    REDIS_PORT: int = 6379
    REDIS_DB: int = 0
    REDIS_PASSWORD: Optional[str] = None
    
    # 测试数据库配置
    TEST_DATABASE_URI: str = "sqlite:///./test.db"
    
    # 日志配置
    LOG_LEVEL: str = "INFO"
    
    # 邮件配置
    SMTP_TLS: bool = True
    SMTP_PORT: Optional[int] = None
    SMTP_HOST: Optional[str] = None
    SMTP_USER: Optional[str] = None
    SMTP_PASSWORD: Optional[str] = None
    EMAILS_FROM_EMAIL: Optional[str] = None
    EMAILS_FROM_NAME: Optional[str] = None
    
    # 超级用户配置
    FIRST_SUPERUSER: str = "admin@example.com"
    FIRST_SUPERUSER_PASSWORD: str = "admin123"
    
    # 项目配置
    PROJECT_NAME: str = "FastAPI Project"
    SENTRY_DSN: Optional[AnyHttpUrl] = None
    
    class Config:
        case_sensitive = True
        env_file = ".env"

settings = Settings()

11.2.2 环境变量文件

# .env
# 数据库配置
POSTGRES_SERVER=localhost
POSTGRES_USER=fastapi_user
POSTGRES_PASSWORD=your_password
POSTGRES_DB=fastapi_db
POSTGRES_PORT=5432

# 安全配置
SECRET_KEY=your-secret-key-here
ACCESS_TOKEN_EXPIRE_MINUTES=43200

# CORS配置
BACKEND_CORS_ORIGINS=["http://localhost:3000","http://localhost:8080"]

# Redis配置
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0

# 邮件配置
SMTP_TLS=True
SMTP_PORT=587
SMTP_HOST=smtp.gmail.com
SMTP_USER=your-email@gmail.com
SMTP_PASSWORD=your-app-password
EMAILS_FROM_EMAIL=your-email@gmail.com
EMAILS_FROM_NAME="FastAPI App"

# 项目配置
PROJECT_NAME="My FastAPI Project"
SERVER_NAME="My FastAPI Server"
SERVER_HOST=http://localhost

# 超级用户配置
FIRST_SUPERUSER=admin@example.com
FIRST_SUPERUSER_PASSWORD=admin123

11.3 数据库连接配置

11.3.1 SQLAlchemy配置

# app/core/database.py
from sqlalchemy import create_engine, MetaData
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from .config import settings
import logging

logger = logging.getLogger(__name__)

# 根据环境选择数据库URI
if settings.SQLALCHEMY_DATABASE_URI:
    SQLALCHEMY_DATABASE_URL = str(settings.SQLALCHEMY_DATABASE_URI)
else:
    SQLALCHEMY_DATABASE_URL = settings.SQLITE_DATABASE_URI

logger.info(f"Database URL: {SQLALCHEMY_DATABASE_URL}")

# 创建数据库引擎
if SQLALCHEMY_DATABASE_URL.startswith("sqlite"):
    # SQLite配置
    engine = create_engine(
        SQLALCHEMY_DATABASE_URL,
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
        echo=settings.LOG_LEVEL == "DEBUG"
    )
else:
    # PostgreSQL/MySQL配置
    engine = create_engine(
        SQLALCHEMY_DATABASE_URL,
        pool_size=settings.DB_POOL_SIZE,
        max_overflow=settings.DB_MAX_OVERFLOW,
        pool_timeout=settings.DB_POOL_TIMEOUT,
        pool_recycle=settings.DB_POOL_RECYCLE,
        echo=settings.LOG_LEVEL == "DEBUG"
    )

# 创建会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建基础模型类
Base = declarative_base()

# 元数据配置
metadata = MetaData(
    naming_convention={
        "ix": "ix_%(column_0_label)s",
        "uq": "uq_%(table_name)s_%(column_0_name)s",
        "ck": "ck_%(table_name)s_%(constraint_name)s",
        "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
        "pk": "pk_%(table_name)s"
    }
)
Base.metadata = metadata

11.3.2 数据库会话管理

# app/db/session.py
from sqlalchemy.orm import Session
from sqlalchemy.exc import SQLAlchemyError
from contextlib import contextmanager
from typing import Generator
from ..core.database import SessionLocal
import logging

logger = logging.getLogger(__name__)

def get_db() -> Generator[Session, None, None]:
    """
    获取数据库会话
    
    Yields:
        Session: 数据库会话
    """
    db = SessionLocal()
    try:
        yield db
    except SQLAlchemyError as e:
        logger.error(f"Database error: {e}")
        db.rollback()
        raise
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        db.rollback()
        raise
    finally:
        db.close()

@contextmanager
def get_db_context() -> Generator[Session, None, None]:
    """
    数据库会话上下文管理器
    
    Yields:
        Session: 数据库会话
    """
    db = SessionLocal()
    try:
        yield db
        db.commit()
    except SQLAlchemyError as e:
        logger.error(f"Database error: {e}")
        db.rollback()
        raise
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        db.rollback()
        raise
    finally:
        db.close()

class DatabaseManager:
    """
    数据库管理器
    """
    
    def __init__(self):
        self.session_factory = SessionLocal
    
    def get_session(self) -> Session:
        """
        获取新的数据库会话
        
        Returns:
            Session: 数据库会话
        """
        return self.session_factory()
    
    @contextmanager
    def session_scope(self) -> Generator[Session, None, None]:
        """
        会话作用域管理器
        
        Yields:
            Session: 数据库会话
        """
        session = self.get_session()
        try:
            yield session
            session.commit()
        except Exception:
            session.rollback()
            raise
        finally:
            session.close()
    
    def execute_transaction(self, func, *args, **kwargs):
        """
        执行事务
        
        Args:
            func: 要执行的函数
            *args: 位置参数
            **kwargs: 关键字参数
            
        Returns:
            函数执行结果
        """
        with self.session_scope() as session:
            return func(session, *args, **kwargs)

# 全局数据库管理器实例
db_manager = DatabaseManager()

11.4 连接池配置

11.4.1 连接池参数详解

# app/core/database_pool.py
from sqlalchemy import create_engine, event
from sqlalchemy.pool import QueuePool, NullPool, StaticPool
from sqlalchemy.engine import Engine
from .config import settings
import logging
import time

logger = logging.getLogger(__name__)

class DatabasePoolConfig:
    """
    数据库连接池配置
    """
    
    @staticmethod
    def create_engine_with_pool(database_url: str, **kwargs):
        """
        创建带连接池的数据库引擎
        
        Args:
            database_url: 数据库连接URL
            **kwargs: 额外参数
            
        Returns:
            Engine: 数据库引擎
        """
        if database_url.startswith("sqlite"):
            # SQLite使用StaticPool
            engine = create_engine(
                database_url,
                poolclass=StaticPool,
                connect_args={"check_same_thread": False},
                echo=kwargs.get("echo", False),
                **kwargs
            )
        else:
            # PostgreSQL/MySQL使用QueuePool
            engine = create_engine(
                database_url,
                poolclass=QueuePool,
                pool_size=settings.DB_POOL_SIZE,
                max_overflow=settings.DB_MAX_OVERFLOW,
                pool_timeout=settings.DB_POOL_TIMEOUT,
                pool_recycle=settings.DB_POOL_RECYCLE,
                pool_pre_ping=True,  # 连接前检查
                echo=kwargs.get("echo", False),
                **kwargs
            )
        
        # 添加连接池事件监听器
        DatabasePoolConfig._setup_pool_events(engine)
        
        return engine
    
    @staticmethod
    def _setup_pool_events(engine: Engine):
        """
        设置连接池事件监听器
        
        Args:
            engine: 数据库引擎
        """
        
        @event.listens_for(engine, "connect")
        def set_sqlite_pragma(dbapi_connection, connection_record):
            """SQLite优化设置"""
            if engine.url.drivername == "sqlite":
                cursor = dbapi_connection.cursor()
                # 启用外键约束
                cursor.execute("PRAGMA foreign_keys=ON")
                # 设置WAL模式
                cursor.execute("PRAGMA journal_mode=WAL")
                # 设置同步模式
                cursor.execute("PRAGMA synchronous=NORMAL")
                cursor.close()
        
        @event.listens_for(engine, "checkout")
        def receive_checkout(dbapi_connection, connection_record, connection_proxy):
            """连接检出事件"""
            connection_record.info['checkout_time'] = time.time()
            logger.debug("Connection checked out from pool")
        
        @event.listens_for(engine, "checkin")
        def receive_checkin(dbapi_connection, connection_record):
            """连接检入事件"""
            checkout_time = connection_record.info.get('checkout_time')
            if checkout_time:
                duration = time.time() - checkout_time
                logger.debug(f"Connection checked in to pool, duration: {duration:.2f}s")
        
        @event.listens_for(engine, "invalidate")
        def receive_invalidate(dbapi_connection, connection_record, exception):
            """连接失效事件"""
            logger.warning(f"Connection invalidated: {exception}")
    
    @staticmethod
    def get_pool_status(engine: Engine) -> dict:
        """
        获取连接池状态
        
        Args:
            engine: 数据库引擎
            
        Returns:
            dict: 连接池状态信息
        """
        pool = engine.pool
        
        return {
            "pool_size": pool.size(),
            "checked_in": pool.checkedin(),
            "checked_out": pool.checkedout(),
            "overflow": pool.overflow(),
            "invalid": pool.invalid()
        }

11.4.2 连接池监控

# app/core/database_monitor.py
from sqlalchemy.engine import Engine
from sqlalchemy import event
from typing import Dict, Any
import time
import threading
from collections import defaultdict, deque
import logging

logger = logging.getLogger(__name__)

class DatabaseMonitor:
    """
    数据库连接监控器
    """
    
    def __init__(self, engine: Engine, max_history: int = 1000):
        self.engine = engine
        self.max_history = max_history
        self.stats = {
            'total_connections': 0,
            'active_connections': 0,
            'failed_connections': 0,
            'total_queries': 0,
            'slow_queries': 0,
            'avg_query_time': 0.0
        }
        self.query_history = deque(maxlen=max_history)
        self.connection_history = deque(maxlen=max_history)
        self._lock = threading.Lock()
        
        self._setup_monitoring()
    
    def _setup_monitoring(self):
        """
        设置监控事件
        """
        
        @event.listens_for(self.engine, "connect")
        def on_connect(dbapi_connection, connection_record):
            with self._lock:
                self.stats['total_connections'] += 1
                self.stats['active_connections'] += 1
                
                self.connection_history.append({
                    'event': 'connect',
                    'timestamp': time.time(),
                    'connection_id': id(dbapi_connection)
                })
        
        @event.listens_for(self.engine, "close")
        def on_close(dbapi_connection, connection_record):
            with self._lock:
                self.stats['active_connections'] = max(0, self.stats['active_connections'] - 1)
                
                self.connection_history.append({
                    'event': 'close',
                    'timestamp': time.time(),
                    'connection_id': id(dbapi_connection)
                })
        
        @event.listens_for(self.engine, "before_cursor_execute")
        def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
            context._query_start_time = time.time()
        
        @event.listens_for(self.engine, "after_cursor_execute")
        def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
            total_time = time.time() - context._query_start_time
            
            with self._lock:
                self.stats['total_queries'] += 1
                
                # 更新平均查询时间
                current_avg = self.stats['avg_query_time']
                total_queries = self.stats['total_queries']
                self.stats['avg_query_time'] = (
                    (current_avg * (total_queries - 1) + total_time) / total_queries
                )
                
                # 记录慢查询（超过1秒）
                if total_time > 1.0:
                    self.stats['slow_queries'] += 1
                    logger.warning(f"Slow query detected: {total_time:.2f}s - {statement[:100]}...")
                
                self.query_history.append({
                    'statement': statement[:200],  # 截断长SQL
                    'duration': total_time,
                    'timestamp': time.time(),
                    'parameters': str(parameters)[:100] if parameters else None
                })
    
    def get_stats(self) -> Dict[str, Any]:
        """
        获取监控统计信息
        
        Returns:
            dict: 统计信息
        """
        with self._lock:
            pool_status = {
                'pool_size': self.engine.pool.size(),
                'checked_in': self.engine.pool.checkedin(),
                'checked_out': self.engine.pool.checkedout(),
                'overflow': self.engine.pool.overflow(),
                'invalid': self.engine.pool.invalid()
            }
            
            return {
                'connection_stats': self.stats.copy(),
                'pool_status': pool_status,
                'recent_queries': list(self.query_history)[-10:],  # 最近10个查询
                'recent_connections': list(self.connection_history)[-10:]  # 最近10个连接事件
            }
    
    def get_slow_queries(self, limit: int = 10) -> list:
        """
        获取慢查询列表
        
        Args:
            limit: 返回数量限制
            
        Returns:
            list: 慢查询列表
        """
        with self._lock:
            slow_queries = [
                query for query in self.query_history 
                if query['duration'] > 1.0
            ]
            return sorted(slow_queries, key=lambda x: x['duration'], reverse=True)[:limit]
    
    def reset_stats(self):
        """
        重置统计信息
        """
        with self._lock:
            self.stats = {
                'total_connections': 0,
                'active_connections': 0,
                'failed_connections': 0,
                'total_queries': 0,
                'slow_queries': 0,
                'avg_query_time': 0.0
            }
            self.query_history.clear()
            self.connection_history.clear()

11.5 数据库初始化

11.5.1 数据库初始化脚本

# app/db/init_db.py
from sqlalchemy.orm import Session
from ..core.database import engine, Base
from ..core.config import settings
from ..models.user import User
from ..crud.user import crud_user
from ..schemas.user import UserCreate
from .session import get_db_context
import logging

logger = logging.getLogger(__name__)

def create_tables():
    """
    创建所有数据库表
    """
    logger.info("Creating database tables...")
    Base.metadata.create_all(bind=engine)
    logger.info("Database tables created successfully")

def init_db() -> None:
    """
    初始化数据库
    """
    # 创建表
    create_tables()
    
    # 创建初始数据
    with get_db_context() as db:
        # 检查是否已存在超级用户
        user = crud_user.get_by_email(db, email=settings.FIRST_SUPERUSER)
        if not user:
            user_in = UserCreate(
                email=settings.FIRST_SUPERUSER,
                password=settings.FIRST_SUPERUSER_PASSWORD,
                is_superuser=True,
                is_active=True,
                full_name="Super User"
            )
            user = crud_user.create(db, obj_in=user_in)
            logger.info(f"Superuser created: {user.email}")
        else:
            logger.info(f"Superuser already exists: {user.email}")

def drop_tables():
    """
    删除所有数据库表
    """
    logger.warning("Dropping all database tables...")
    Base.metadata.drop_all(bind=engine)
    logger.warning("All database tables dropped")

def reset_db():
    """
    重置数据库
    """
    drop_tables()
    init_db()
    logger.info("Database reset completed")

if __name__ == "__main__":
    init_db()

11.5.2 数据库健康检查

# app/core/database_health.py
from sqlalchemy import text
from sqlalchemy.orm import Session
from sqlalchemy.exc import SQLAlchemyError
from ..db.session import get_db_context
from typing import Dict, Any
import time
import logging

logger = logging.getLogger(__name__)

class DatabaseHealthChecker:
    """
    数据库健康检查器
    """
    
    @staticmethod
    def check_connection() -> Dict[str, Any]:
        """
        检查数据库连接
        
        Returns:
            dict: 健康检查结果
        """
        start_time = time.time()
        
        try:
            with get_db_context() as db:
                # 执行简单查询
                result = db.execute(text("SELECT 1"))
                result.fetchone()
                
                response_time = time.time() - start_time
                
                return {
                    "status": "healthy",
                    "response_time": round(response_time * 1000, 2),  # 毫秒
                    "timestamp": time.time(),
                    "message": "Database connection successful"
                }
                
        except SQLAlchemyError as e:
            response_time = time.time() - start_time
            logger.error(f"Database health check failed: {e}")
            
            return {
                "status": "unhealthy",
                "response_time": round(response_time * 1000, 2),
                "timestamp": time.time(),
                "error": str(e),
                "message": "Database connection failed"
            }
        
        except Exception as e:
            response_time = time.time() - start_time
            logger.error(f"Unexpected error during health check: {e}")
            
            return {
                "status": "error",
                "response_time": round(response_time * 1000, 2),
                "timestamp": time.time(),
                "error": str(e),
                "message": "Unexpected error occurred"
            }
    
    @staticmethod
    def check_tables() -> Dict[str, Any]:
        """
        检查数据库表
        
        Returns:
            dict: 表检查结果
        """
        try:
            with get_db_context() as db:
                # 检查用户表是否存在
                result = db.execute(text(
                    "SELECT name FROM sqlite_master WHERE type='table' AND name='users'"
                    if "sqlite" in str(db.bind.url)
                    else "SELECT tablename FROM pg_tables WHERE tablename='users'"
                ))
                
                tables_exist = result.fetchone() is not None
                
                return {
                    "status": "healthy" if tables_exist else "warning",
                    "tables_exist": tables_exist,
                    "timestamp": time.time(),
                    "message": "Tables exist" if tables_exist else "Tables not found"
                }
                
        except Exception as e:
            logger.error(f"Table check failed: {e}")
            return {
                "status": "error",
                "tables_exist": False,
                "timestamp": time.time(),
                "error": str(e),
                "message": "Table check failed"
            }
    
    @staticmethod
    def get_database_info() -> Dict[str, Any]:
        """
        获取数据库信息
        
        Returns:
            dict: 数据库信息
        """
        try:
            with get_db_context() as db:
                # 获取数据库版本
                if "sqlite" in str(db.bind.url):
                    version_query = "SELECT sqlite_version()"
                elif "postgresql" in str(db.bind.url):
                    version_query = "SELECT version()"
                else:
                    version_query = "SELECT @@version"
                
                result = db.execute(text(version_query))
                version = result.fetchone()[0]
                
                return {
                    "database_type": str(db.bind.url).split("://")[0],
                    "version": version,
                    "url": str(db.bind.url).split("@")[-1] if "@" in str(db.bind.url) else str(db.bind.url),
                    "timestamp": time.time()
                }
                
        except Exception as e:
            logger.error(f"Failed to get database info: {e}")
            return {
                "error": str(e),
                "timestamp": time.time()
            }

# 全局健康检查器实例
db_health_checker = DatabaseHealthChecker()

实践练习

练习1：企业级数据库架构设计

目标：设计一个支持多数据库、读写分离的企业级架构

要求：

• 配置PostgreSQL主从复制
• 集成Redis缓存层
• 实现数据库路由和负载均衡
• 添加连接池监控和自动故障转移

验收标准：

• 支持读写分离，读操作自动路由到从库
• 缓存命中率达到80%以上
• 数据库连接池利用率保持在70%以下
• 故障转移时间小于5秒

练习2：高性能数据访问层

目标：构建高性能的数据访问层和查询优化系统

要求：

• 实现智能查询缓存
• 添加SQL查询分析和优化建议
• 集成数据库连接池监控
• 实现批量操作优化

验收标准：

• 查询响应时间减少50%以上
• 支持10000+并发数据库连接
• 自动识别慢查询并提供优化建议
• 批量操作性能提升3倍以上

练习3：数据库迁移与版本管理系统

目标：建立完整的数据库版本控制和自动化迁移系统

要求：

• 集成Alembic实现数据库迁移
• 构建迁移脚本管理平台
• 实现零停机迁移策略
• 添加迁移回滚和数据备份机制

验收标准：

• 支持自动化数据库迁移部署
• 迁移过程可视化监控
• 支持一键回滚到任意版本
• 迁移过程数据零丢失

常见问题

Q1: 数据库连接池配置过大导致资源浪费

问题描述：连接池配置不当，导致数据库资源浪费或连接不足

解决方案：

# 动态连接池配置
class DynamicConnectionPool:
    def __init__(self):
        self.min_connections = 5
        self.max_connections = 50
        self.current_load = 0
    
    def adjust_pool_size(self, current_connections: int, active_connections: int):
        """根据负载动态调整连接池大小"""
        utilization = active_connections / current_connections if current_connections > 0 else 0
        
        if utilization > 0.8:  # 高负载，增加连接
            new_size = min(current_connections + 5, self.max_connections)
        elif utilization < 0.3:  # 低负载，减少连接
            new_size = max(current_connections - 2, self.min_connections)
        else:
            new_size = current_connections
        
        return new_size

Q2: SQLAlchemy查询性能优化

问题描述：复杂查询性能差，N+1查询问题频发

解决方案：

# 查询优化策略
class OptimizedQuery:
    @staticmethod
    def get_users_with_posts(db: Session, limit: int = 100):
        """优化的用户和文章查询"""
        return db.query(User).options(
            joinedload(User.posts),  # 预加载关联数据
            selectinload(User.profile)  # 选择性加载
        ).limit(limit).all()
    
    @staticmethod
    def bulk_update_users(db: Session, updates: List[Dict]):
        """批量更新优化"""
        db.bulk_update_mappings(User, updates)
        db.commit()

Q3: 数据库迁移失败和回滚

问题描述：数据库迁移过程中出现错误，需要安全回滚

解决方案：

# 安全迁移策略
class SafeMigration:
    def __init__(self, db_url: str):
        self.db_url = db_url
        self.backup_path = None
    
    def create_backup(self) -> str:
        """创建迁移前备份"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        backup_file = f"backup_{timestamp}.sql"
        
        # 执行备份命令
        subprocess.run([
            "pg_dump", self.db_url, "-f", backup_file
        ], check=True)
        
        self.backup_path = backup_file
        return backup_file
    
    def migrate_with_rollback(self, migration_script: str):
        """带回滚的安全迁移"""
        try:
            # 创建备份
            self.create_backup()
            
            # 执行迁移
            alembic.command.upgrade(alembic_cfg, "head")
            
        except Exception as e:
            logger.error(f"Migration failed: {e}")
            # 自动回滚
            self.rollback_from_backup()
            raise

总结

本章深入介绍了FastAPI中的数据库连接与配置，主要内容包括：

• 数据库集成基础：掌握了SQLAlchemy与FastAPI的集成方式，理解了ORM的核心概念
• 连接管理优化：学习了数据库连接池配置、会话管理和连接优化策略
• 模型设计实践：实现了完整的数据模型设计，包括关系映射和约束定义
• CRUD操作封装：构建了可复用的数据访问层，提高了代码的可维护性
• 性能优化技巧：掌握了查询优化、索引设计和缓存策略的最佳实践
• 健康监控机制：实现了数据库健康检查和性能监控系统

通过本章学习，您已经具备了构建企业级数据库应用的核心技能。

下一步

• 学习 012-SQLAlchemy集成：深入了解ORM框架集成和使用
• 探索数据库高级特性：分区表、物化视图、全文搜索等
• 研究微服务架构下的数据库设计模式

参考资源

• SQLAlchemy官方文档 - SQLAlchemy完整指南
• FastAPI数据库教程 - 官方数据库集成教程
• Alembic文档 - 数据库迁移工具
• PostgreSQL性能调优 - 数据库性能优化
• 数据库设计模式 - 企业应用架构模式

更新记录

• 更新时间: 2024-12-19 | 更新内容: 完善数据库连接与配置教程，添加实践练习和常见问题解答 | 更新人: Assistant

1. 数据库配置管理的最佳实践
2. SQLAlchemy集成和连接配置
3. 连接池管理和性能优化
4. 数据库监控和健康检查
5. 环境配置和安全管理
6. 数据库初始化和维护

正确的数据库配置是构建高性能、可靠Web应用的基础。在下一章中，我们将深入学习SQLAlchemy ORM的使用，包括模型定义、关系映射等核心概念。