> ## Documentation Index
> Fetch the complete documentation index at: https://docs.akria.net/llms.txt
> Use this file to discover all available pages before exploring further.
# Day 20 | 第三阶段综合实战
> 构建运维数据查询 API 实战：集成 AsyncSQL + Pydantic + Docker End-to-End 开发
**实战日**：Day 20 是 Python Web 阶段的"毕业大考"。我们将把 Week 3 所有的知识点串联起来，构建一个名为 `InsightOps-API` 的微服务。今天不只是写代码，而是要彻底理解 **"如何设计一个完整的 API 项目"**、**"如何组织项目结构"** 以及 **"如何让代码具备生产级质量"**。
## 实战目标 | Project Goal
开发一个用于查询服务器监控数据的 RESTful API。
* **POST /servers**: 注册新服务器（自动校验 IP 格式）
* **GET /servers**: 分页查询服务器列表
* **GET /servers/{id}**: 查询特定服务器详情
* **PUT /servers/{id}**: 更新服务器信息
* **DELETE /servers/{id}**: 删除服务器
* **Web 框架**: FastAPI
* **数据验证**: Pydantic
* **数据库**: SQLAlchemy (Async) + MySQL
* **部署**: Docker + Docker Compose
* **配置管理**: pydantic-settings
***
## 代码任务 (120 mins)
确保所有依赖已安装：
```bash theme={null}
# 确保虚拟环境已激活
source .venv/bin/activate
# 安装所有依赖
pip install fastapi uvicorn gunicorn \
pydantic-settings python-dotenv \
sqlalchemy aiomysql
# 确保 MySQL 服务正在运行
# docker run -d -p 3306:3306 --name mysql-learn -e MYSQL_ROOT_PASSWORD=root mysql:8.0
```
一个标准的 Python 后端项目结构：
```text theme={null}
insightops_api/
├── app/
│ ├── __init__.py
│ ├── main.py # 入口文件
│ ├── models.py # 数据库模型 (ORM)
│ ├── schemas.py # Pydantic 模型 (Validation)
│ ├── database.py # 数据库连接
│ └── config.py # 配置读取
├── requirements.txt
├── Dockerfile
├── docker-compose.yaml
└── .env # 环境变量（不提交到 Git）
```
**为什么这样组织？**
* **模块化**：每个文件负责一个功能，便于维护
* **可扩展**：易于添加新功能（如认证、缓存）
* **标准化**：符合 Python 项目的最佳实践
按照项目结构，逐步实现各个模块。
```python app/config.py theme={null}
#!/usr/bin/env python3
"""配置管理"""
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
app_name: str = "InsightOps API"
debug: bool = False
database_url: str
class Config:
env_file = ".env"
case_sensitive = False
settings = Settings()
```
```python app/database.py theme={null}
#!/usr/bin/env python3
"""数据库配置"""
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from app.config import settings
# 创建异步引擎
engine = create_async_engine(
settings.database_url,
echo=True,
pool_pre_ping=True
)
# 创建会话工厂
AsyncSessionLocal = sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False
)
# 声明基类
Base = declarative_base()
```
```python app/models.py theme={null}
#!/usr/bin/env python3
"""数据库模型"""
from sqlalchemy import Column, Integer, String, Float
from app.database import Base
class Server(Base):
__tablename__ = "servers"
id = Column(Integer, primary_key=True, index=True)
hostname = Column(String(50), unique=True, nullable=False)
ip_address = Column(String(20), nullable=False)
region = Column(String(20), default="us-east-1")
cpu_cores = Column(Integer, nullable=False)
memory_gb = Column(Float, nullable=False)
```
```python app/schemas.py theme={null}
#!/usr/bin/env python3
"""Pydantic 模型"""
from pydantic import BaseModel, IPvAnyAddress, Field
class ServerCreate(BaseModel):
hostname: str = Field(..., min_length=3)
ip_address: IPvAnyAddress # 自动校验 IP 格式
region: str = "us-east-1"
cpu_cores: int = Field(..., gt=0)
memory_gb: float = Field(..., gt=0)
class ServerResponse(ServerCreate):
id: int
class Config:
from_attributes = True
```
```python app/main.py theme={null}
#!/usr/bin/env python3
"""FastAPI 应用入口"""
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.future import select
from sqlalchemy.ext.asyncio import AsyncSession
from app import models, schemas, database, config
app = FastAPI(
title=config.settings.app_name,
version="1.0.0"
)
# 启动时自动建表
@app.on_event("startup")
async def startup():
async with database.engine.begin() as conn:
await conn.run_sync(models.Base.metadata.create_all)
# 数据库依赖
async def get_db():
async with database.AsyncSessionLocal() as session:
yield session
@app.post("/servers/", response_model=schemas.ServerResponse, status_code=201)
async def create_server(
server: schemas.ServerCreate,
db: AsyncSession = Depends(get_db)
):
# 检查 Hostname 是否重复
result = await db.execute(
select(models.Server).where(models.Server.hostname == server.hostname)
)
if result.scalars().first():
raise HTTPException(status_code=400, detail="Hostname already exists")
# 创建服务器对象
db_server = models.Server(
hostname=server.hostname,
ip_address=str(server.ip_address),
region=server.region,
cpu_cores=server.cpu_cores,
memory_gb=server.memory_gb
)
db.add(db_server)
await db.commit()
await db.refresh(db_server)
return db_server
@app.get("/servers/", response_model=list[schemas.ServerResponse])
async def list_servers(
skip: int = 0,
limit: int = 10,
db: AsyncSession = Depends(get_db)
):
result = await db.execute(
select(models.Server).offset(skip).limit(limit)
)
return result.scalars().all()
@app.get("/servers/{server_id}", response_model=schemas.ServerResponse)
async def get_server(
server_id: int,
db: AsyncSession = Depends(get_db)
):
result = await db.execute(
select(models.Server).where(models.Server.id == server_id)
)
server = result.scalars().first()
if not server:
raise HTTPException(status_code=404, detail="Server not found")
return server
```
```yaml docker-compose.yaml theme={null}
version: '3.8'
services:
db:
image: mysql:8.0
environment:
MYSQL_ROOT_PASSWORD: root
MYSQL_DATABASE: insight_ops
ports:
- "3306:3306"
volumes:
- mysql_data:/var/lib/mysql
api:
build: .
command: uvicorn app.main:app --host 0.0.0.0 --port 8000
ports:
- "8000:8000"
environment:
DATABASE_URL: mysql+aiomysql://root:root@db/insight_ops
depends_on:
- db
restart: unless-stopped
volumes:
mysql_data:
```
**项目结构说明**：
* **`app/`**：应用主目录
* **`models.py`**：数据库模型（ORM）
* **`schemas.py`**：Pydantic 模型（验证）
* **`database.py`**：数据库连接配置
* **`config.py`**：配置管理
* **`main.py`**：FastAPI 应用入口
**运行项目**：
```bash theme={null}
# 使用 Docker Compose 启动
docker-compose up --build
# 或者本地运行
uvicorn app.main:app --reload
```
**验证步骤**：
1. 访问 `http://localhost:8000/docs` 打开 Swagger UI
2. 测试创建服务器：
* 点击 `POST /servers/`
* 输入数据，执行，应该返回 201
3. 测试查询服务器列表：
* 点击 `GET /servers/`
* 执行，应该看到服务器列表
4. 测试查询单个服务器：
* 点击 `GET /servers/{server_id}`
* 输入服务器 ID，执行，应该看到服务器详情
***
## 拓展任务 (30 mins)
**任务**：为 API 添加 Token 认证，只有认证用户才能创建服务器。
**提示**：使用 Day 17 学到的依赖注入实现。
**任务**：添加请求日志中间件，记录每个请求的详细信息。
**提示**：使用 Day 17 学到的中间件实现。
***
## 今日产出物
* 完整的 `InsightOps-API` 项目
* Docker 镜像和 Docker Compose 配置
* 生产级 API 服务
***
## 参考代码
在 GitHub 查看完整的 InsightOps API 项目
包含完整的 FastAPI + SQLAlchemy + Docker 实现
查看标准项目目录结构
学习生产级项目组织方式
***
## 实际应用场景
* **微服务架构**：构建轻量级微服务
* **API 网关**：作为 API 网关，统一管理接口
* **数据查询服务**：提供数据查询和操作接口
* **自动化工具**：为运维工具提供 RESTful API
* **监控系统**：提供监控数据的查询接口
* **模块化设计**：按功能拆分模块，便于维护
* **配置外置**：使用环境变量管理配置
* **容器化部署**：使用 Docker 实现可移植性
* **文档完善**：利用 FastAPI 自动生成文档
* **错误处理**：统一的错误处理和响应格式
**与 Day 21 的关联**：今天完成的综合项目，明天会进行复盘和总结，回顾第三周学到的所有知识点。
***
Day 19 | 生产部署与配置
Day 21 | 第三阶段复盘