配置和数据库连接

企业项目里，数据库地址、JWT 密钥、运行端口都不应该写死在代码里。

这一节实现：

app/core/config.py：统一配置管理。
app/db/base.py：SQLAlchemy ORM 基类。
app/db/session.py：数据库引擎、会话、连接检查。
lifespan：应用启动时检查数据库连接。

一、配置管理 config.py

app/core/config.py：

from functools import lru_cache

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
from sqlalchemy.engine import make_url


class Settings(BaseSettings):
    app_name: str = Field(default="my-fastapi", validation_alias="APP_NAME")
    app_env: str = Field(default="local", validation_alias="APP_ENV")
    app_host: str = Field(default="127.0.0.1", validation_alias="APP_HOST")
    app_port: int = Field(default=8000, validation_alias="APP_PORT")
    debug: bool = Field(default=True, validation_alias="APP_DEBUG")
    db_echo: bool = Field(default=False, validation_alias="DB_ECHO")

    # 敏感配置不写默认值，必须从 .env 或系统环境变量提供。
    database_url: str = Field(validation_alias="DATABASE_URL")
    jwt_secret_key: str = Field(validation_alias="JWT_SECRET_KEY")
    jwt_algorithm: str = Field(default="HS256", validation_alias="JWT_ALGORITHM")
    jwt_access_token_expire_minutes: int = Field(
        default=1440,
        validation_alias="JWT_ACCESS_TOKEN_EXPIRE_MINUTES",
    )

    # 鉴权白名单，支持以 * 结尾的前缀匹配。
    auth_exclude_paths: list[str] = Field(
        default_factory=lambda: [
            "/",
            "/docs*",
            "/redoc*",
            "/openapi.json",
            "/favicon.ico",
            "/api/health",
            "/api/auth/register",
            "/api/auth/login",
        ],
        validation_alias="AUTH_EXCLUDE_PATHS",
    )

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
    )


@lru_cache
def get_settings() -> Settings:
    # 缓存配置对象，避免重复读取 .env。
    return Settings()


settings = get_settings()


def get_database_url_info() -> dict[str, str | int | None]:
    # 打印日志时只输出脱敏连接信息，禁止泄露密码。
    url = make_url(settings.database_url)
    return {
        "driver": url.drivername,
        "username": url.username,
        "host": url.host,
        "port": url.port,
        "database": url.database,
    }

关键点：

写法	说明
`validation_alias="APP_NAME"`	Python 字段名和环境变量名分开
`database_url` 不给默认值	避免生产环境忘配数据库
`jwt_secret_key` 不给默认值	避免代码里写死密钥
`@lru_cache`	配置只初始化一次
`make_url()`	安全解析数据库 URL，打印时不输出密码

二、定义 ORM Base

app/db/base.py：

from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass

所有 SQLAlchemy 模型都继承这个 Base。

三、创建数据库会话

app/db/session.py：

from collections.abc import Generator

from sqlalchemy import create_engine, text
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session, sessionmaker

from app.core.config import get_database_url_info, settings

engine = create_engine(
    settings.database_url,
    echo=settings.db_echo,
    pool_pre_ping=True,
    pool_recycle=3600,
)

SessionLocal = sessionmaker(
    bind=engine,
    class_=Session,
    autoflush=False,
    autocommit=False,
)


def get_db() -> Generator[Session, None, None]:
    # 每个请求创建一个独立 Session，请求结束后关闭。
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


def check_database_connection() -> bool:
    # 启动时做一次轻量检查，方便快速发现数据库配置问题。
    try:
        with engine.connect() as connection:
            connection.execute(text("SELECT 1"))
    except SQLAlchemyError as exc:
        print("数据库连接失败:", get_database_url_info(), exc)
        return False

    print("数据库连接正常:", get_database_url_info())
    return True

参数说明：

参数	说明
`pool_pre_ping=True`	每次取连接前先检查连接是否可用
`pool_recycle=3600`	定期回收连接，减少 MySQL 连接超时问题
`autoflush=False`	不自动 flush，代码行为更清晰
`autocommit=False`	手动 `commit()` 控制事务

四、启动时检查数据库

app/main.py：

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.core.config import settings
from app.db.session import check_database_connection


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # 启动时检查数据库连接。是否阻止启动，可以按项目要求决定。
    check_database_connection()
    yield


def create_app() -> FastAPI:
    app = FastAPI(
        title=settings.app_name,
        debug=settings.debug,
        lifespan=lifespan,
    )

    @app.get("/")
    def index():
        return {"name": settings.app_name, "env": settings.app_env}

    return app


app = create_app()

FastAPI 当前推荐用 lifespan 管理启动和关闭逻辑，不建议新项目继续使用旧的 @app.on_event("startup")。

五、不要自动建表

企业项目建议不要在启动时写：

Base.metadata.create_all(bind=engine)

原因：

表结构变更不可追踪。
生产环境容易误改表。
多人协作时不知道数据库当前版本。

正确做法是使用 Alembic 管理迁移，后面单独讲。

#配置和数据库连接

#一、配置管理 config.py

#二、定义 ORM Base

#三、创建数据库会话

#四、启动时检查数据库