FastAPI构建Python微服务指南（python微服务开发）

以下是使用FastAPI构建Python微服务的分步指南，结合“粉丝福利社”场景的示例：

1. 环境准备

bash

# 创建虚拟环境

python -m venv venv

source venv/bin/activate # Linux/Mac

venv\Scripts\activate.bat # Windows

# 安装依赖

pip install fastapi uvicorn[standard] sqlalchemy asyncpg pydantic-settings

2. 项目结构

fan-service/

├── src/

│ ├── core/

│ │ ├── config.py # 配置管理

│ ├── models/

│ │ ├── schemas.py # Pydantic模型

│ │ └── database.py # 数据库模型

│ ├── services/

│ │ ├── user_service.py # 用户服务

│ │ └── welfare_service.py # 福利服务

│ └── main.py # 主入口

├── .env # 环境变量

└── requirements.txt

3. 配置管理 (config.py)

python

from pydantic_settings import BaseSettings

class Settings(BaseSettings):

DATABASE_URL: str = "postgresql+asyncpg://user:pass@localhost:5432/fan_db"

JWT_SECRET: str = "your-secret-key"

class Config:

env_file = ".env"

settings = Settings()

4. 数据库模型 (database.py)

python

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()

engine = create_async_engine(settings.DATABASE_URL)

async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db():

async with async_session() as session:

yield session

5. 用户服务示例 (user_service.py)

python

from fastapi import APIRouter, Depends, HTTPException

from sqlalchemy.ext.asyncio import AsyncSession

from .models import UserCreate, UserResponse

from .database import get_db

router = APIRouter(prefix="/users", tags=["users"])

@router.post("/", response_model=UserResponse)

async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):

# 密码哈希处理

# 数据库存储

return {...}

6. 福利服务示例 (welfare_service.py)

python

from fastapi import APIRouter, Depends

from sqlalchemy.ext.asyncio import AsyncSession

from .schemas import WelfareCreate, WelfareResponse

from .database import get_db

router = APIRouter(prefix="/welfares", tags=["welfares"])

@router.post("/", response_model=WelfareResponse)

async def create_welfare(welfare: WelfareCreate,

db: AsyncSession = Depends(get_db)):

# 创建福利项目逻辑

# 库存管理

# 发放限制

return {...}

7. 主入口 (main.py)

python

from fastapi import FastAPI

from .services import user_service, welfare_service

from .core.config import settings

app = FastAPI(title="粉丝福利社", version="1.0.0")

app.include_router(user_service.router)

app.include_router(welfare_service.router)

@app.get("/health")

async def health_check():

return {"status": "healthy"}

if __name__ == "__main__":

import uvicorn

uvicorn.run(app, host="0.0.0.0", port=8000)

8. 运行与测试

bash

# 启动服务

uvicorn src.main:app --reload

# 测试API

curl
http://localhost:8000/docs # 查看Swagger文档

关键功能扩展

用户认证（JWT示例）

python

from fastapi.security import OAuth2PasswordBearer

from jose import JWTError, jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):

credentials_exception = HTTPException(

status_code=401,

detail="Invalid credentials"

)

try:

payload = jwt.decode(token, settings.JWT_SECRET, algorithms=["HS256"])

return payload

except JWTError:

raise credentials_exception

异步数据库操作

python

from sqlalchemy import select

from .models import User

async def get_user_by_email(email: str, db: AsyncSession):

result = await db.execute(select(User).filter(User.email == email))

return result.scalar_one_or_none()

微服务通信（HTTPX示例）

python

import httpx

async def notify_service(user_id: int, message: str):

async with httpx.AsyncClient() as client:

response = await client.post(

"http://notification-service/notify",

json={"user_id": user_id, "message": message}

)

response.raise_for_status()

部署建议

1. 容器化（Docker示例）：

dockerfile

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "80"]

2. 性能优化：

• 使用Gunicorn多进程：

bash

gunicorn -w 4 -k uvicorn.workers.UvicornWorker src.main:app

3. 监控：

• 集成Prometheus：

python

from fastapi import FastAPI

from prometheus_fastapi_instrumentator import Instrumentator

app = FastAPI()

Instrumentator().instrument(app).expose(app)

最佳实践

1. 使用依赖注入管理资源
2. 采用领域驱动设计（DDD）划分微服务边界
3. 实现Circuit Breaker模式增强容错能力
4. 使用API Gateway进行路由和聚合
5. 为每个服务独立数据库
6. 实施自动化测试（单元测试、集成测试）
7. 配置完善的日志和监控系统

这个架构可以支持典型的粉丝福利社功能：

• 用户身份验证和权限管理
• 福利商品管理
• 福利领取和发放
• 用户行为追踪
• 实时通知推送
• 数据统计分析

根据具体业务需求，可以继续扩展积分系统、抽奖功能、活动限时模块等。