后端开发指南

环境准备

cd apps/backend

# 创建虚拟环境
python -m venv .venv
source .venv/bin/activate  # Linux/macOS
# .venv\Scripts\activate   # Windows

# 安装依赖
pip install -r requirements.txt

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

目录结构

apps/backend/
├── app/
│   ├── api/              # REST API 路由
│   │   ├── auth.py       # 认证接口
│   │   ├── projects.py   # 项目接口
│   │   ├── training.py   # 训练任务接口
│   │   ├── admin.py      # 管理后台接口
│   │   └── internal.py   # 内部服务接口
│   ├── core/
│   │   ├── config.py     # 配置管理（pydantic-settings）
│   │   ├── security.py   # JWT 认证
│   │   └── deps.py       # 依赖注入
│   ├── models/           # SQLAlchemy 数据模型
│   ├── schemas/          # Pydantic 请求/响应模型
│   ├── services/         # 业务逻辑层
│   ├── tasks/            # Celery 任务定义
│   ├── agent/            # 节点监控 Agent
│   └── main.py           # FastAPI 应用入口
├── alembic/              # 数据库迁移
├── tests/                # 测试文件
├── Dockerfile
└── Dockerfile.migrate

API 开发

添加新路由

1. 在 app/api/ 创建或编辑路由文件
2. 在 app/main.py 注册路由

# app/api/example.py
from fastapi import APIRouter, Depends

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

@router.get("/")
async def list_items():
    return {"items": []}

请求/响应模型

使用 Pydantic 定义数据模型：

# app/schemas/example.py
from pydantic import BaseModel

class ItemCreate(BaseModel):
    name: str
    description: str | None = None

class ItemResponse(BaseModel):
    id: int
    name: str
    description: str | None

业务逻辑

在 app/services/ 中实现业务逻辑：

# app/services/example.py
from app.models import Item
from app.schemas import ItemCreate

class ItemService:
    async def create_item(self, data: ItemCreate) -> Item:
        # 业务逻辑
        pass

Celery 任务

定义任务

# app/tasks/example.py
from app.core.celery_app import celery_app

@celery_app.task(queue='gpu')
def train_model(project_id: int, params: dict):
    # 训练逻辑
    pass

提交任务

from app.tasks.example import train_model

# 异步提交
result = train_model.delay(project_id=1, params={...})

# 获取结果
result.get(timeout=3600)

配置管理

使用 pydantic-settings 管理配置：

# app/core/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    postgres_host: str = "postgres"
    postgres_port: int = 5432
    postgres_db: str = "radstudio"
    secret_key: str = "change-me"

    class Config:
        env_file = ".env"

常用命令

uvicorn app.main:app --reload    # 启动开发服务器
pytest                           # 运行测试
pytest -m no_db                  # 跳过数据库测试
alembic upgrade head             # 应用迁移
alembic revision --autogenerate -m "desc"  # 创建迁移