模多多教育版

Appearance

模多多教育版 – 后端项目框架 (FastAPI + SQLAlchemy 2.0) ​

本文档提供 V1.0 版本的后端工程脚手架。目标：

  1. 快速启动本地开发环境；2. 提供统一、清晰的目录结构；3. 约定 API / Service / Model / Schema 规范；4. 对接前端同事所需接口；5. 便于 CI/CD 与后期扩展。

1. 技术栈 & 运行时 ​

领域选型
语言Python 3.11
Web 框架FastAPI (Starlette ASGI)
ORMSQLAlchemy 2.0 (async) + Alembic 迁移
依赖管理Poetry or pip + requirements.txt
数据库PostgreSQL 17
缓存 / 队列Redis 7
消息WebSocket (Starlette) / RabbitMQ (可选)
AIOpenAI / 腾讯 TRTC SDK (WebRTC)
容器Docker + Docker Compose

2. 顶层目录结构 ​

text
moduoduo-backend/
├── app/
│   ├── api/                # 路由层：HTTP ↔ Service
│   │   ├── __init__.py
│   │   ├── auth.py
│   │   ├── courses.py
│   │   ├── tasks.py
│   │   ├── ai.py
│   │   └── billing.py
│   ├── core/               # 基础设施：配置 / 安全 / 依赖
│   │   ├── __init__.py
│   │   ├── config.py
│   │   ├── security.py
│   │   └── dependencies.py
│   ├── models/             # SQLAlchemy ORM 实体
│   │   ├── __init__.py
│   │   ├── user.py
│   │   ├── course.py
│   │   ├── lesson.py
│   │   ├── task.py
│   │   ├── assignment.py
│   │   ├── submission.py
│   │   ├── ai_session.py
│   │   ├── token.py
│   │   └── base.py         # Base = declarative_base()
│   ├── schemas/            # Pydantic 请求 / 响应
│   │   ├── __init__.py
│   │   ├── user.py
│   │   ├── course.py
│   │   ├── task.py
│   │   ├── ai.py
│   │   └── billing.py
│   ├── services/           # 业务逻辑层
│   │   ├── __init__.py
│   │   ├── ai_service.py
│   │   ├── billing_service.py
│   │   └── live_service.py
│   ├── db/
│   │   ├── __init__.py
│   │   ├── session.py      # async engine / session maker
│   │   └── init_db.py      # 初始数据脚本
│   └── main.py             # ASGI 入口
├── alembic/                # 数据库迁移
│   ├── env.py
│   └── versions/
├── tests/                  # Pytest
├── migrations.sh           # 快捷脚本
├── Dockerfile
├── docker-compose.yml      # 本地 Postgres + Redis
├── pyproject.toml          # (Poetry) 或 requirements.txt
├── .env.example            # 环境变量模板
└── README.md

3. 依赖安装 & 本地启动 ​

bash
# 1) 克隆仓库
$ git clone xxx.git && cd moduoduo-backend

# 2) 安装依赖
$ pip install -r requirements.txt   # 或 poetry install

# 3) 复制环境变量模板
$ cp .env.example .env

# 4) 启动服务（需本地 Postgres/Redis，可用 docker compose）
$ docker-compose up -d pg redis
$ alembic upgrade head  # 初始化数据库
$ uvicorn app.main:app --reload

TIP: docker-compose up -d 同时会起 pgadmin4 方便可视化查看数据。

4. 核心代码示例 ​

4.1 app/main.py ​

python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.api import auth, courses, tasks, ai, billing
from app.core.config import settings

app = FastAPI(
    title="模多多教育版 API",
    version="1.0.0",
    openapi_url="/openapi.json",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.ALLOW_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 路由注册
app.include_router(auth.router, prefix="/auth", tags=["Auth"])
app.include_router(courses.router, prefix="/courses", tags=["Courses"])
app.include_router(tasks.router, prefix="/tasks", tags=["Tasks"])
app.include_router(ai.router, prefix="/ai", tags=["AI"])
app.include_router(billing.router, prefix="/billing", tags=["Billing"])

4.2 app/core/config.py ​

python
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    # 数据库
    DATABASE_URL: str = "postgresql+asyncpg://user:pass@localhost:5432/moduoduo"

    # 安全
    SECRET_KEY: str = "changeme"  # 🔒 请替换
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24

    # CORS
    ALLOW_ORIGINS: list[str] = ["*"]

    model_config = SettingsConfigDict(env_file=".env", case_sensitive=True)

settings = Settings()

4.3 app/db/session.py ​

python
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from app.core.config import settings

engine = create_async_engine(settings.DATABASE_URL, echo=False, pool_pre_ping=True)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

async def get_db() -> AsyncSession:  # FastAPI Depends
    async with AsyncSessionLocal() as session:
        yield session

4.4 app/models/user.py ​

python
from sqlalchemy import String, Enum, Integer, ForeignKey, DateTime
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.models.base import Base
import enum, datetime as dt

class UserRole(str, enum.Enum):
    STUDENT = "student"
    TEACHER = "teacher"
    SCHOOL_ADMIN = "school_admin"
    PLATFORM_ADMIN = "platform_admin"

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    username: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
    password_hash: Mapped[str] = mapped_column(String(128), nullable=False)
    role: Mapped[UserRole] = mapped_column(Enum(UserRole), default=UserRole.STUDENT)
    school_id: Mapped[int | None] = mapped_column(ForeignKey("schools.id"))
    real_name: Mapped[str | None] = mapped_column(String(64))
    created_at: Mapped[dt.datetime] = mapped_column(DateTime, default=dt.datetime.utcnow)

    # relationships
    school = relationship("School", back_populates="users")

4.5 app/schemas/user.py ​

python
from pydantic import BaseModel, Field
from enum import Enum

class UserRole(str, Enum):
    student = "student"
    teacher = "teacher"
    school_admin = "school_admin"
    platform_admin = "platform_admin"

class UserCreate(BaseModel):
    username: str = Field(min_length=3, max_length=64)
    password: str = Field(min_length=6)
    role: UserRole = UserRole.student

class UserRead(BaseModel):
    id: int
    username: str
    role: UserRole

    class Config:
        from_attributes = True

4.6 app/api/courses.py (示例部分路由) ​

python
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.schemas import course as schema
from app.services.course_service import CourseService
from app.db.session import get_db

router = APIRouter()

@router.get("", response_model=list[schema.CourseListItem])
async def list_courses(student_id: int, db: AsyncSession = Depends(get_db)):
    return await CourseService(db).list_for_student(student_id)

@router.get("/{course_id}", response_model=schema.CourseDetail)
async def get_course(course_id: int, db: AsyncSession = Depends(get_db)):
    course = await CourseService(db).get(course_id)
    if not course:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Course not found")
    return course

4.7 app/services/ai_service.py (片段) ​

python
class AIService:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def generate_drawing(self, *, user_id: int, task_id: int, prompt: str) -> str:
        # 1) 计算 tokens 消耗 & 扣费
        # 2) 调用第三方 AI 绘图
        # 3) 记录 ai_sessions
        # 4) 返回 image_url
        ...

5. API 约定 & 自动文档 ​

  • 路径：全部复数名词 (RESTful)；示例 /courses/{id}/publish。
  • 鉴权：Authorization: Bearer <JWT>，登陆接口 /auth/login 返回 access_token。
  • 错误格式：
json
{ "code": 404, "message": "Not Found", "detail": null }
  • FastAPI 自动生成 Swagger UI，访问 http://localhost:8000/docs。

6. ALEMBIC 数据迁移 ​

bash
# 生成迁移文件
$ alembic revision -m "create users & courses"
# 应用迁移
$ alembic upgrade head

在 alembic/env.py 中引入 app.models.base.Base.metadata。

7. 测试 & CI 模板 ​

yaml
# .github/workflows/ci.yml
name: CI
on: [push]
jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_USER: user
          POSTGRES_PASSWORD: pass
          POSTGRES_DB: moduoduo_test
        ports: ["5432:5432"]
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: "3.11"
      - run: pip install -r requirements.txt
      - run: pytest -q

8. 后续扩展建议 ​

  1. 领域分层：保持 API ↔ Service ↔ Repository (DB) 解耦。
  2. AI 调用限流 & 熔断：可用 redis + asyncio.Semaphore 控制并发；集成熔断器库如 aiobreaker。
  3. 事件队列：复杂任务（大文件上传转码、AI 批量评测）建议推入 Celery/RabbitMQ 队列。
  4. 监控 & 日志：接入 prometheus-fastapi-instrumentator 和 structlog。
  5. 多租户 (多学校)：在 settings 支持 schema_name 或 Row-Level Security。

💡 至此，你已获得可运行的后端骨架。根据前端联调需求，依照 app/api/* 和 app/services/* 继续迭代实现具体业务逻辑。

9. Docker 一键启动 & 常见问题 ​

适用环境：Windows (Docker Desktop + WSL 2 或 Hyper-V)、macOS、Linux。

9.1 安装前置 ​

  1. 安装 Docker Desktop
  2. 启用 CPU 虚拟化、WSL 2 (Windows Home) 或 Hyper-V (Windows Pro)
  3. 打开终端验证：docker version、docker compose version

9.2 创建 .env ​

在仓库根目录新建 .env（与 docker-compose.yml 同级），拷贝下方内容并按需修改：

bash
DATABASE_URL=postgresql+asyncpg://moduoduo:moduoduo@postgres:5432/moduoduo
SECRET_KEY=ReplaceThisWithRandom
ACCESS_TOKEN_EXPIRE_MINUTES=1440
ALLOW_ORIGINS=*

9.3 一键启动 ​

bash
docker compose up --build -d  # 首次启动（含构建）

将创建 3 个容器：

  • postgres – PostgreSQL 17 (5432)
  • redis – Redis 7 (6379)
  • backend – FastAPI (8000)

9.4 初始化数据库 ​

bash
docker compose exec backend alembic upgrade head

9.5 热更新（开发模式） ​

bash
# 停掉后台 backend 容器
docker compose stop backend
# 前台运行, 代码保存即重载
docker compose run --service-ports backend \
  uvicorn app.main:app --reload --host 0.0.0.0

9.6 重新构建镜像 ​

  • 依赖或 Dockerfile 改动 → docker compose build backend
  • 仅代码改动 → docker compose up -d 会复用缓存

9.7 常见报错速查 ​

报错日志处理办法
.env not found根目录补上 .env 文件
version is obsolete可忽略或删掉 version: 字段
No matching distribution更新 requirements.txt 到可用版本，如 prometheus-fastapi-instrumentator==7.1.0
SettingsError ALLOW_ORIGINS已在 app/core/config.py 加入字符串→列表转换；确保 .env 字段正确

10. 下一步开发路线 ​

  1. 定义数据模型 – 在 app/models/ 按数据库设计补充表；alembic revision --autogenerate 生成迁移脚本。
  2. 实现业务服务 – 在 app/services/ 编写核心逻辑（DB + AI 调用）。
  3. 开放 API – 在 app/api/ 编写路由，使用 Pydantic schemas/ 进行参数 / 响应校验。
  4. 单元 / 集成测试 – 用 pytest + FastAPI TestClient；CI 已集成。
  5. 前端联调 – 把 /openapi.json 提供给前端，或直接访问 /docs 查看接口说明。
  6. 代码质量 – 可加入 ruff/black 格式化、pre-commit 钩子。

至此，后端基础设施完整可跑；后续迭代只需遵循 模型 → 迁移 → 服务 → API → 测试 流程即可。