投稿
  1. 其其网首页
  2. 投稿

API 路由组织方法详解（AI 视频下载总结器的 RESTful 设计）

拓扑大师 • • 投稿

后台接口从十几个涨到六十多个时，全堆在 main.py 里的路由让改一处支付回调都要滚动半天。我后来按”业务域”把路由拆进独立的 Router 模块，统一挂在 /api 前缀下，再靠依赖注入把鉴权和数据库会话收敛到一处，整个接口层才变得清爽，新功能从提需求到上线不用再为路径写法争论。下面就是这套路由设计的完整落地过程，包含目录结构、注册方式、RESTful 约定以及我们踩过的坑。

一、整体结构

路由一旦多起来，最先受不了的就是单个文件。main.py 里同时塞着认证、视频、总结、支付、用户的接口，改一个文件要小心别动到别人的函数，code review 时 diff 又大又难读。我决定按业务域拆：认证、视频、总结、问答、支付、用户、管理各占一个文件，每个文件只关心自己领域内的接口，main.py 退化成纯粹的挂载点。

app/
├── main.py              # 入口，挂载所有 router
├── api/
│   ├── __init__.py
│   ├── deps.py          # 公共依赖
│   ├── auth.py          # 认证
│   ├── video.py         # 视频相关
│   ├── summary.py       # 总结
│   ├── qa.py            # 问答
│   ├── payment.py       # 支付
│   ├── user.py          # 用户
│   └── admin.py         # 管理

目录定下来之后，团队协作的边界也清晰了：一个人负责视频域，另一个人改支付域，几乎不会冲突。init.py 让 api 目录成为包，main.py 里 import 时路径一致，也从根上避免了循环导入的问题。

二、Router 注册

模块拆好了，下一步是把它们挂到应用上。这里有两个关键点：一是统一加 /api 前缀，让所有业务接口共享一个入口，Nginx 转发、跨域配置、日志过滤都只用写一次；二是给每个 router 打 tags，OpenAPI 文档里会按标签分组，前端同学查接口方便很多。

# main.py
from fastapi import FastAPI
from app.api import auth, video, summary, qa, payment, user, admin

app = FastAPI(title='AI 万能视频下载总结器', version='1.0.0')

# 统一前缀 /api
app.include_router(auth.router, prefix='/api/auth', tags=['认证'])
app.include_router(video.router, prefix='/api/video', tags=['视频'])
app.include_router(summary.router, prefix='/api/summary', tags=['总结'])
app.include_router(qa.router, prefix='/api/qa', tags=['问答'])
app.include_router(payment.router, prefix='/api/payment', tags=['支付'])
app.include_router(user.router, prefix='/api/user', tags=['用户'])
app.include_router(admin.router, prefix='/api/admin', tags=['管理'])

这里有个易错点：prefix 只在 include_router 里写一次，router 内部定义路径时不要再带 /api/video 这种前缀，否则会出现 /api/video/api/video 的重复路径。我们团队早期就在这上面栽过，后来约定成”路由文件里只写相对路径”。

三、RESTful 设计原则

路径怎么命名，我们内部讨论过一轮。有同事习惯动作式命名（/createVideo、/deleteVideo），但这类命名把动作塞进 URL，客户端一多就不好扩展。我最后定下一套规则：资源用复数名词，动作交给 HTTP 方法表达，下面是平台的接口对照表。

资源 路径 方法
视频 /api/video POST 创建（提交链接）
/api/video/{id} GET 详情
/api/video/{id} DELETE 删除
/api/video GET 列表
总结 /api/summary POST 创建（同步）
/api/summary/stream POST 流式
/api/summary/{id} GET 详情
问答 /api/qa POST 提问
/api/qa/history/{summary_id} GET 历史
支付 /api/payment/create-session POST
/api/payment/webhook POST
用户 /api/user/me GET
/api/user/quota GET
/api/user/subscription GET

这张表基本覆盖了平台的全部操作。坚持这套约定后，前端把路径和方法一一对应，几乎不需要文档之外的沟通；create-session 和 webhook 这类动作形态的接口单独用动词，属于例外处理，但都集中在支付域，不会扩散到其他模块。

四、单个 Router 范例

以 video.py 为例，一个文件承载视频资源的完整生命周期：提交链接、查详情、列列表、删除。核心思路是接口只做参数校验和数据编排，真正的下载逻辑放到 service 层，避免路由函数里堆业务代码。

# api/video.py
from fastapi import APIRouter, Depends, HTTPException, Query
from typing import Optional

from app.api.deps import get_current_user, get_db
from app.models import User, Video
from app.schemas.video import (
    VideoCreate, VideoInfo, VideoList
)
from app.services.downloader import VideoDownloader
from sqlalchemy.ext.asyncio import AsyncSession

router = APIRouter()

@router.post('', response_model=VideoInfo, summary='提交视频链接')
async def create_video(
    req: VideoCreate,
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    """提交视频链接，获取元信息（不下载）"""
    downloader = VideoDownloader()
    info = await downloader.get_info(req.url)
    
    video = Video(
        user_id=user.id,
        url=req.url,
        platform=info.platform,
        video_id=info.video_id,
        title=info.title,
        thumbnail=info.thumbnail,
        duration=info.duration,
        uploader=info.uploader,
        status='pending'
    )
    db.add(video)
    await db.commit()
    await db.refresh(video)
    return video

@router.get('/{video_id}', response_model=VideoInfo, summary='视频详情')
async def get_video(
    video_id: int,
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    video = await db.get(Video, video_id)
    if not video or video.user_id != user.id:
        raise HTTPException(404, "Video not found")
    return video

@router.get('', response_model=VideoList, summary='视频列表')
async def list_videos(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    offset = (page - 1) * page_size
    result = await db.execute(
        select(Video)
        .where(Video.user_id == user.id, Video.is_deleted == False)
        .order_by(Video.created_at.desc())
        .offset(offset)
        .limit(page_size)
    )
    videos = result.scalars().all()
    total = await db.scalar(
        select(func.count(Video.id))
        .where(Video.user_id == user.id, Video.is_deleted == False)
    )
    return {'items': videos, 'total': total, 'page': page, 'page_size': page_size}

@router.delete('/{video_id}', summary='删除视频')
async def delete_video(
    video_id: int,
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    video = await db.get(Video, video_id)
    if not video or video.user_id != user.id:
        raise HTTPException(404, "Video not found")
    
    video.is_deleted = True
    # 同时删除物理文件
    if video.video_path:
        os.remove(video.video_path)
    
    await db.commit()
    return {'message': 'Deleted'}

这段代码解决了”接口逻辑散落”的问题：提交链接时只抓取元信息不下载，视频状态置为 pending，由后台任务真正拉取文件，前端拿到 200 就认为提交成功。注意列表接口做了软删除，is_deleted 置真而不是物理删行，这样误删还能找回，审计也有记录；删除视频时再同步清理磁盘文件，避免存储泄漏。

五、依赖注入

接口一多，每个函数都要重复写”取当前用户、开数据库会话”这段样板，删减一点都会引起连锁修改。我把它们抽成 FastAPI 依赖，函数签名里声明 Depends 就能自动注入，业务函数只保留自己的参数。

# api/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer
from sqlalchemy.ext.asyncio import AsyncSession
from jose import JWTError, jwt

from app.db import async_session
from app.config import settings
from app.models import User

security = HTTPBearer()

async def get_db() -> AsyncSession:
    """数据库 session 依赖"""
    async with async_session() as session:
        yield session

async def get_current_user(
    credentials = Depends(security),
    db: AsyncSession = Depends(get_db)
) -> User:
    """当前用户依赖"""
    token = credentials.credentials
    try:
        payload = jwt.decode(
            token, settings.JWT_SECRET, algorithms=['HS256']
        )
        user_id = int(payload['sub'])
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail='Invalid token'
        )
    
    user = await db.get(User, user_id)
    if not user or not user.is_active:
        raise HTTPException(401, 'User not found')
    
    return user

async def get_admin_user(
    user: User = Depends(get_current_user)
) -> User:
    """管理员用户依赖"""
    if not user.is_admin:
        raise HTTPException(403, 'Admin only')
    return user

依赖注入之后，鉴权逻辑只维护一份，改签发算法、加 token 黑名单都在这一处。之前鉴权散落在各接口里，改一次要动十几个文件；现在新增接口只需要在签名里写 Depends(getcurrentuser)，几秒钟就能挂上鉴权。

六、Schema 验证

请求参数和响应结构全部走 Pydantic，非法输入在进入业务逻辑之前就被拦截。视频链接这种字段，自己写正则很容易漏掉各种域名变体，直接声明成 HttpUrl 类型让库去校验，省心又可靠。

# schemas/video.py
from pydantic import BaseModel, Field, HttpUrl
from datetime import datetime
from typing import Optional

class VideoCreate(BaseModel):
    url: HttpUrl = Field(..., description='视频链接')
    download_video: bool = Field(False, description='是否下载视频文件')
    
    class Config:
        json_schema_extra = {
            'example': {
                'url': 'https://www.youtube.com/watch?v=dQw4w9WgXcQ',
                'download_video': False
            }
        }

class VideoInfo(BaseModel):
    id: int
    url: str
    platform: str
    title: Optional[str]
    thumbnail: Optional[str]
    duration: int
    uploader: Optional[str]
    status: str
    created_at: datetime
    
    class Config:
        from_attributes = True  # SQLAlchemy 兼容

class VideoList(BaseModel):
    items: list[VideoInfo]
    total: int
    page: int
    page_size: int

VideoCreate 里的 example 配置会显示在 Swagger UI 中，前端可以直接点”Try it out”联调。from_attributes 让 ORM 对象能直接转成响应模型，省去手写转换函数。响应模型一旦定下来，OpenAPI 文档里的返回结构也跟着固定，前后端不会各说各话。

七、错误处理

业务错误如果用默认的 HTTPException，返回体只有固定的 detail 字段，前端还得自己解析文案，判断逻辑怎么写都别扭。我自定义了 APIError 异常，统一返回 {code, message} 结构，前端拿到 code 做判断，文案直接展示。

# main.py
from fastapi import Request
from fastapi.responses import JSONResponse

class APIError(Exception):
    def __init__(self, code: str, message: str, status: int = 400):
        self.code = code
        self.message = message
        self.status = status

@app.exception_handler(APIError)
async def api_error_handler(request: Request, exc: APIError):
    return JSONResponse(
        status_code=exc.status,
        content={'code': exc.code, 'message': exc.message}
    )

业务代码里：

if not video:
    raise APIError('VIDEO_NOT_FOUND', '视频不存在', 404)

if user.daily_used >= user.daily_video_quota:
    raise APIError('QUOTA_EXCEEDED', '今日次数已用完', 429)

统一异常处理后，前端只需要处理 {code, message} 一种结构，配额、限流、资源不存在都是同一种交互方式。429 状态码还方便客户端感知”今天次数用完了”，从而引导用户去升级会员，而不是在界面上干转圈。

八、中间件

跨域、压缩、请求日志这些横切关注点，如果每个接口自己处理，代码会非常啰嗦。把它们放在中间件里，业务代码完全不用感知，只管自己的参数和返回值。

# main.py
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware

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

# Gzip
app.add_middleware(GZipMiddleware, minimum_size=1000)

# 请求日志
@app.middleware('http')
async def log_requests(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    
    logger.info(f'{request.method} {request.url.path} '
                f'{response.status_code} {duration:.3f}s')
    return response

CORS 的来源必须从配置读取，别用 * 放行，否则 Cookie 和鉴权头都带不过去。日志中间件记录了每个请求的方法、路径、状态码和耗时，排查”哪个接口慢”直接翻日志按耗时排序，比反复问前端快得多。

九、限流

提交视频链接的接口被脚本刷过几次，同一个 IP 一分钟内打了几百次，把下载队列占满，正常用户反而排不上。我加了一层限流保护，把异常流量挡在业务逻辑外面。

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@router.post('')
@limiter.limit('5/minute')
async def create_video(request: Request, ...):
    ...

单机部署下 slowapi 够用，每分钟 5 次对正常用户是宽松的，对脚本是明确的门槛。等以后集群化部署，再换成 Redis 计数，这里保留替换空间，限流逻辑不用推倒重来。

十、版本管理

版本策略我们犹豫过是放 URL 还是放 Header，最后选了 URL，直观且前端不用额外配置。v1 时期接口还不太稳定，老接口先留着 deprecated 标记，给迁移留出时间窗口。

# 未来 /api/v2 兼容老版
app.include_router(auth.router, prefix='/api/v1/auth', tags=['认证'])

老版本标记 deprecated：

@router.post('', deprecated=True)
async def old_login():
    """v1 接口，已废弃，请使用 /api/auth/login"""
    ...

deprecated 标记会让 OpenAPI 文档里这个接口变灰，前端在 Swagger 上一眼就能看到该换新接口了。URL 版本号保证老客户端在过渡期内不挂，新老版本可以并行跑一段时间再下线。

十一、OpenAPI 文档

FastAPI 根据路由和 Schema 自动生成 OpenAPI 文档，这是选它做接口层的一个现实收益，文档零维护成本。

/docs      # Swagger UI
/redoc     # ReDoc
/openapi.json  # OpenAPI spec

平台在前端用 openapi-typescript 生成类型：

npx openapi-typescript http://localhost:8000/openapi.json -o types.ts

前端从 openapi.json 生成 TypeScript 类型之后，后端改了响应结构，前端构建期就直接报类型错误，比运行时才发现问题省事得多。契约由代码驱动，而不是靠口头约定，两边对齐的成本几乎为零。

十二、API 设计细节

这一部分是我们定下来、并且一直遵守的几个小约定，它们共同保证了路径风格统一，新接口照着写就不会跑偏。

1. URL 用复数

✅ /api/videos
❌ /api/video

2. 不用动词

✅ POST /api/videos    （创建视频）
❌ POST /api/createVideo

3. 用 HTTP method 表达动作

GET /api/videos        # 列表
GET /api/videos/{id}   # 详情
POST /api/videos       # 创建
PUT /api/videos/{id}   # 全量更新
PATCH /api/videos/{id} # 部分更新
DELETE /api/videos/{id} # 删除

4. 复杂操作用子资源

POST /api/videos/{id}/summarize
POST /api/summaries/{id}/qa
POST /api/payments/refund

这些约定看着细碎，但正是它们让接口可以被前端”猜”出来：看到 /api/videos/{id} 就知道能 GET 也能 DELETE，看到子资源就知道它依赖父资源。复杂动作放在子资源上，避免在 URL 里发明动词，也让路由规则保持简单一致。

十三、踩过的坑

挑几个真实影响过开发效率的坑记录在这里，团队新成员看了能少走弯路，也当作复盘留档。

  • 前缀重复：路由已经 prefix=’/api’，方法里又写 ‘/api/auth’。统一在一处写。
  • 依赖循环：A router 引用 B router 的 service，B 引用 A。重构为 service 独立。
  • async 传染：同步函数调阻塞 IO 会卡事件循环。所有 IO 都要 async。
  • 响应模型缺失：不写 response_model 返回的是 dict，OpenAPI 不规范。
  • HTTP 状态码错：用 200 表示”逻辑错误”（应该 400/422）。FastAPI 用 HTTPException。
  • CORS preflight 慢：OPTIONS 请求要快返回。配置 CORS 允许方法。
  • 大响应不流式：1MB+ JSON 响应慢。用 StreamingResponse。
  • 认证信息泄露：错误信息不要说”用户不存在 vs 密码错误”，会泄露用户存在性。

前缀重复和依赖循环属于结构问题，越早发现代价越低；状态码和响应模型属于规范问题，靠 code review 兜底。最后一条关于认证信息泄露，是安全评审时发现的，注册接口提示”邮箱已存在”同理，统一返回模糊的错误信息，不给攻击者留枚举口子。

十四、监控

接口上了生产之后，光有日志不够，还要有指标才能量化体验。我接入了 Prometheus 客户端，用 Counter 和 Histogram 分别统计请求计数与延迟分布。

from prometheus_client import Counter, Histogram

REQUEST_COUNT = Counter('http_requests_total', 
    'Total HTTP requests',
    ['method', 'endpoint', 'status'])
REQUEST_LATENCY = Histogram('http_request_duration_seconds',
    'HTTP request latency',
    ['method', 'endpoint'])

@app.middleware('http')
async def monitor(request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    
    REQUEST_COUNT.labels(
        method=request.method,
        endpoint=request.url.path,
        status=response.status_code
    ).inc()
    REQUEST_LATENCY.labels(
        method=request.method,
        endpoint=request.url.path
    ).observe(duration)
    
    return response

这个监控中间件和前面的日志中间件可以共存，一个出指标一个出明细。Grafana 上配一个 95 分位延迟的面板，哪个域慢了立刻能看见，比用户反馈更早暴露问题。

十五、最佳实践清单

把前面所有经验收敛成一张清单，作为接口设计的验收标准，每条都能在 code review 时逐项打勾。

  1. 按业务域分 Router；
  2. 统一 /api/{domain} 前缀；
  3. RESTful URL 设计；
  4. 依赖注入鉴权/DB；
  5. Pydantic Schema 校验；
  6. 统一异常处理；
  7. OpenAPI 自动文档；
  8. Prometheus 监控；
  9. 限流防滥用；
  10. 错误信息脱敏。

到这里，这套路由从目录结构到监控告警的落地链路就完整了。照着清单走，新接口从提出到上线不需要再争论路径该怎么写，团队可以集中精力处理真正的业务逻辑。

常见问题（FAQ）

Q1：要不要用 GraphQL？

单端用户场景 REST 够用。GraphQL 适合”客户端驱动”复杂查询。

Q2：版本管理用 URL 还是 Header？

URL 简单直观，平台用 URL 版本。

Q3：API 限流用 Redis 还是中间件？

Redis 集群方案。中间件（如 slowapi）单机够用。

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 qiqicto@qq.com 举报，一经查实，本站将立刻删除。
赞 (0)
拓扑大师的头像拓扑大师普通用户
0

相关推荐

返回顶部