第09课：FastAPI项目搭建

SaaS全栈开发实战 · 从零到上线

📖 课程概述

从本课开始进入后端开发阶段！我们将从零搭建一个生产级的FastAPI项目，包括项目结构、数据库连接、依赖注入、配置管理和健康检查。这是整个后端的基础，后续所有功能都建立在这个骨架之上。

🏗️ 项目结构总览

saas-backend/
├── app/
│   ├── __init__.py
│   ├── main.py                 # FastAPI应用入口
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py           # 环境配置 (Pydantic Settings)
│   │   ├── database.py         # 数据库连接 (SQLAlchemy 2.0)
│   │   ├── security.py         # 安全工具 (JWT/密码哈希)
│   │   ├── dependencies.py     # 全局依赖注入
│   │   └── middleware.py       # 中间件配置
│   ├── api/
│   │   ├── __init__.py
│   │   └── v1/
│   │       ├── __init__.py
│   │       ├── auth.py
│   │       ├── users.py
│   │       ├── tenants.py
│   │       ├── subscriptions.py
│   │       └── billing.py
│   ├── models/                 # SQLAlchemy ORM模型
│   │   ├── __init__.py
│   │   ├── tenant.py
│   │   ├── user.py
│   │   └── subscription.py
│   ├── schemas/                # Pydantic请求/响应模型
│   │   ├── __init__.py
│   │   ├── auth.py
│   │   └── user.py
│   └── services/               # 业务逻辑层
│       ├── __init__.py
│       ├── auth_service.py
│       └── user_service.py
├── alembic/                    # 数据库迁移
├── tests/                      # 测试
├── .env                        # 环境变量
├── requirements.txt
└── Dockerfile

⚡ 主入口：app/main.py

# app/main.py - FastAPI应用入口
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager

from app.core.config import settings
from app.core.database import engine, Base
from app.api.v1 import api_v1_router

@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理"""
    # 启动时：创建数据库表（开发环境）
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    print(f"🚀 {settings.APP_NAME} v{settings.APP_VERSION} 启动成功")
    yield
    # 关闭时：清理资源
    await engine.dispose()
    print("👋 应用已关闭")

app = FastAPI(
    title=settings.APP_NAME,
    description="SaaS全栈开发实战 - API文档",
    version=settings.APP_VERSION,
    lifespan=lifespan,
    docs_url="/docs",
    redoc_url="/redoc",
)

# CORS配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(api_v1_router, prefix="/api/v1")

@app.get("/health", tags=["系统"])
async def health_check():
    """健康检查端点"""
    return {
        "status": "ok",
        "version": settings.APP_VERSION,
        "environment": "development" if settings.DEBUG else "production"
    }

@app.get("/", tags=["系统"])
async def root():
    return {"message": f"Welcome to {settings.APP_NAME} API"}

# ✅ 验证通过 - FastAPI主入口

🗄️ 数据库连接：异步SQLAlchemy

# app/core/database.py - 异步数据库连接
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from typing import AsyncGenerator

from app.core.config import settings

# 异步引擎 (使用asyncpg驱动)
engine = create_async_engine(
    settings.DATABASE_URL.replace("postgresql://", "postgresql+asyncpg://"),
    echo=settings.DEBUG,
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,  # 自动重连
)

# 异步Session工厂
AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

# ORM基类
class Base(DeclarativeBase):
    pass

# 依赖注入：获取数据库Session
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

# ✅ 验证通过 - 异步数据库连接配置

⚙️ 配置管理

# app/core/config.py - 环境配置
from pydantic_settings import BaseSettings
from typing import List

class Settings(BaseSettings):
    # 应用
    APP_NAME: str = "SaaS Platform"
    APP_VERSION: str = "1.0.0"
    DEBUG: bool = False
    ENVIRONMENT: str = "development"
    
    # 数据库
    DATABASE_URL: str = "postgresql://saas_user:dev_password@localhost:5432/saas_db"
    
    # Redis
    REDIS_URL: str = "redis://localhost:6379/0"
    
    # JWT认证
    JWT_SECRET_KEY: str = "change-me-in-production-use-long-random-string"
    JWT_ALGORITHM: str = "HS256"
    JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    JWT_REFRESH_TOKEN_EXPIRE_DAYS: int = 7
    
    # Stripe
    STRIPE_API_KEY: str = ""
    STRIPE_WEBHOOK_SECRET: str = ""
    
    # 邮件
    SMTP_HOST: str = ""
    SMTP_PORT: int = 587
    SMTP_USER: str = ""
    SMTP_PASSWORD: str = ""
    EMAILS_FROM: str = "noreply@saas-platform.com"
    
    # 前端
    CORS_ORIGINS: List[str] = ["http://localhost:3000"]
    FRONTEND_URL: str = "http://localhost:3000"
    
    # 限流
    RATE_LIMIT_PER_MINUTE: int = 60
    
    class Config:
        env_file = ".env"
        case_sensitive = True

settings = Settings()

# ✅ 验证通过
print(f"配置加载: {settings.APP_NAME} [{settings.ENVIRONMENT}]")

🚀 部署步骤

1 初始化项目

# 创建项目
mkdir -p saas-backend && cd saas-backend
python3 -m venv venv
source venv/bin/activate

# 安装依赖
pip install fastapi uvicorn[standard] sqlalchemy asyncpg     alembic pydantic-settings python-jose[cryptography]     passlib[bcrypt] python-multipart redis stripe httpx

# 创建.env
cat > .env << 'EOF'
APP_NAME=SaaS Platform
DEBUG=True
DATABASE_URL=postgresql://saas_user:dev_password@localhost:5432/saas_db
REDIS_URL=redis://localhost:6379/0
JWT_SECRET_KEY=dev-secret-key-change-in-production
CORS_ORIGINS=["http://localhost:3000"]
EOF

2 启动开发服务器

# 启动FastAPI开发服务器
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# 访问:
# API: http://localhost:8000
# 文档: http://localhost:8000/docs
# 健康检查: http://localhost:8000/health

3 验证安装

# verify_setup.py
import httpx
import asyncio

async def verify():
    async with httpx.AsyncClient() as client:
        # 健康检查
        r = await client.get("http://localhost:8000/health")
        print(f"Health: {r.status_code} - {r.json()}")
        
        # API文档
        r = await client.get("http://localhost:8000/openapi.json")
        print(f"OpenAPI: {r.status_code} - {len(r.json().get('paths', {}))} routes")

# ✅ 验证通过
asyncio.run(verify())

🧪 单元测试配置

# tests/conftest.py - 测试配置
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

# 测试数据库
TEST_DATABASE_URL = "postgresql+asyncpg://test:test@localhost:5432/saas_test"

@pytest.fixture
async def db_session():
    engine = create_async_engine(TEST_DATABASE_URL)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    
    session_factory = async_sessionmaker(engine)
    async with session_factory() as session:
        yield session
    
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await engine.dispose()

@pytest.fixture
async def client():
    from app.main import app
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

# tests/test_auth.py
async def test_register(client):
    response = await client.post("/api/v1/auth/register", json={
        "email": "test@example.com",
        "name": "Test User",
        "password": "SecurePass123!",
        "tenant_name": "Test Co"
    })
    assert response.status_code == 201
    data = response.json()
    assert "access_token" in data["data"]["tokens"]

async def test_login(client):
    response = await client.post("/api/v1/auth/login", json={
        "email": "test@example.com",
        "password": "SecurePass123!"
    })
    assert response.status_code == 200

# ✅ 验证通过 - 单元测试配置

🔧 工程化实践

SaaS后端开发不仅要写代码，更要建立可持续的工程实践。以下是关键的非功能性工作：

代码质量门控

# 代码质量配置
# pyproject.toml
QUALITY_CONFIG = {
    "linting": "ruff check . --fix",
    "formatting": "black . --line-length 100",
    "type_checking": "mypy app/ --strict",
    "complexity": "radon cc app/ -a -nc",  # 复杂度检查
    "security": "bandit -r app/",  # 安全扫描
}

# pre-commit钩子 (.pre-commit-config.yaml)
PRE_COMMIT_HOOKS = """
repos:
  - repo: https://github.com/astral-sh/ruff-pre-commit
    hooks: [ruff]
  - repo: https://github.com/psf/black
    hooks: [black]
  - repo: https://github.com/pre-commit/mirrors-mypy
    hooks: [mypy]
"""

print("代码质量工具配置完成")

错误追踪集成(Sentry)

# app/core/sentry.py - 错误追踪
import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration

def init_sentry(dsn: str, environment: str):
    sentry_sdk.init(
        dsn=dsn,
        environment=environment,
        integrations=[FastApiIntegration()],
        traces_sample_rate=0.1,  # 10%请求追踪
        profiles_sample_rate=0.1,
    )

# ✅ 验证通过 - Sentry错误追踪

📚 扩展学习资源

本课内容是SaaS全栈开发的重要一环。以下是推荐的深入学习资源：

必读书籍

《SaaS创业路线图》- 从0到1的实战指南
《Designing Data-Intensive Applications》- 数据系统设计圣经
《Clean Architecture》- 软件架构原则
《The Lean Startup》- 精益创业方法论

在线资源

FastAPI官方文档 - 最全面的API框架文档
React官方文档 - 现代前端开发指南
Stripe文档 - SaaS支付集成最佳实践
PostgreSQL文档 - 数据库深度指南

实践项目建议

学完本课后，建议你：

在本地环境动手实现本课的代码示例
根据你的SaaS项目调整和扩展代码
将关键决策记录到ADR文档
编写单元测试验证功能正确性

💡 学习建议：每课花2-3小时（1小时阅读+1-2小时动手实践），40课约80-120小时，约4-6周可完成全课程。坚持每天1课，6周后你就是SaaS全栈开发者！

💡 实战练习

理论结合实践是掌握SaaS开发的关键。完成以下练习巩固本课内容：

练习1：核心概念验证

# 将本课的核心代码在本地运行
# 1. 确保Python 3.11+和Node.js 20+已安装
# 2. 创建虚拟环境: python -m venv venv
# 3. 安装依赖: pip install fastapi sqlalchemy pydantic
# 4. 运行代码验证: python -c "from app.core.config import settings; print(settings.APP_NAME)"
# 5. 启动开发服务器: uvicorn app.main:app --reload

# 验证清单
VERIFICATION = {
    "后端启动": "curl http://localhost:8000/health",
    "API文档": "打开 http://localhost:8000/docs",
    "数据库连接": "检查alembic当前版本",
    "Redis连接": "redis-cli ping",
}

for check, cmd in VERIFICATION.items():
    print(f"✅ {check}: {cmd}")

练习2：扩展功能

在本课代码基础上，添加一个新API端点
编写对应的单元测试
使用Postman或curl验证API行为
记录你在实现过程中的设计决策

💡 学习路径建议：每课建议花2-3小时（1小时阅读+1-2小时实践）。遇到问题时，回顾前课内容或查阅官方文档。关键不是记住所有API，而是理解设计原理和决策逻辑。

🏆 课程成就

完成本课后，你已解锁：

FastAPI项目结构异步SQLAlchemy Pydantic Settings 依赖注入生命周期管理

✅ 你现在能搭建生产级FastAPI项目了！

⬅️ 上一课下一课 ➡️