Fastapi——GET、POST、图片视频请求




2020-03-02

blog_main_img

GET 查询、POST JSON、图片上传、视频上传、图片返回、视频播放，再补一点 Python 客户端请求示例

FastAPI 写接口的手感很直接：函数参数就是请求参数，Pydantic 模型就是 JSON Body，文件上传交给 UploadFile，文件返回交给 FileResponse 或 StreamingResponse。

先搭一个最小服务

安装常用依赖：

pip install fastapi "uvicorn[standard]" python-multipart

python-multipart 是处理表单和文件上传需要的包。只写 JSON 接口时用不上，一旦接口里有 File 或 Form，它就该进场。

新建 main.py：

from fastapi import FastAPI

app = FastAPI(title="FastAPI Request Demo")


@app.get("/")
async def index():
    return {"message": "FastAPI is ready"}

启动服务：

uvicorn main:app --reload --host 127.0.0.1 --port 8000

浏览器打开：

http://127.0.0.1:8000/docs

FastAPI 会自动生成接口调试页面，GET、POST、文件上传都能在页面里试。

GET：路径参数和查询参数

GET 常用于查询。路径里带的是资源定位，问号后面带的是筛选条件。

FastAPI 路由和请求参数

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/{item_id}")
async def get_item(
    item_id: int,
    q: str | None = Query(default=None, max_length=50),
    page: int = Query(default=1, ge=1),
    size: int = Query(default=10, ge=1, le=100),
    available: bool | None = None,
):
    return {
        "item_id": item_id,
        "q": q,
        "page": page,
        "size": size,
        "available": available,
    }

访问：

GET /items/100?q=keyboard&page=1&size=20&available=true

FastAPI 会自动做几件事：

  • item_id 会被转换成 int
  • page 和 size 会按规则校验
  • available=true 会被转换成 bool
  • 参数不合法时直接返回结构化错误

这种写法很适合列表查询、详情查询、搜索接口。

POST：JSON Body 交给 Pydantic

POST 常用于新增、提交、复杂查询。Body 里的 JSON 用 Pydantic 模型接收最舒服。

from pydantic import BaseModel, Field


class ItemIn(BaseModel):
    name: str = Field(min_length=1, max_length=60)
    price: float = Field(gt=0)
    tags: list[str] = []
    description: str | None = None


class ItemOut(ItemIn):
    id: int


STORE: dict[int, ItemOut] = {}


@app.post("/items", response_model=ItemOut)
async def create_item(item: ItemIn):
    item_id = len(STORE) + 1
    saved = ItemOut(id=item_id, **item.model_dump())
    STORE[item_id] = saved
    return saved

请求体：

{
  "name": "mechanical keyboard",
  "price": 399.0,
  "tags": ["office", "input"],
  "description": "clean layout"
}

这里的重点是 ItemIn。它不只是接收字段，还负责校验字段类型、长度、范围。接口越多，Pydantic 模型越能减少重复判断。

POST 也能接查询参数

别把 POST 和 JSON Body 绑死。POST 一样可以接查询参数。

@app.post("/items/search")
async def search_items(keyword: str, item: ItemIn):
    return {
        "keyword": keyword,
        "condition": item,
    }

请求形态像这样：

POST /items/search?keyword=keyboard
Content-Type: application/json

Body 仍然是 JSON，keyword 仍然来自查询参数。FastAPI 会按参数来源自动拆开。

图片上传：用 UploadFile

图片上传属于 multipart/form-data。FastAPI 里推荐用 UploadFile，它有文件名、内容类型和异步读取方法。

FastAPI 图片视频上传流程

from pathlib import Path
from uuid import uuid4

from fastapi import File, HTTPException, UploadFile


MEDIA_DIR = Path("media")
MEDIA_DIR.mkdir(exist_ok=True)

IMAGE_TYPES = {
    "image/jpeg": ".jpg",
    "image/png": ".png",
    "image/webp": ".webp",
}


@app.post("/images")
async def upload_image(file: UploadFile = File(...)):
    suffix = IMAGE_TYPES.get(file.content_type or "")

    if suffix is None:
        raise HTTPException(status_code=400, detail="只支持 jpg、png、webp 图片")

    target = MEDIA_DIR / f"{uuid4().hex}{suffix}"
    content = await file.read()
    target.write_bytes(content)

    return {
        "filename": target.name,
        "content_type": file.content_type,
        "size": len(content),
        "url": f"/images/{target.name}",
    }

这段代码做了三件事：

  • 校验 content_type
  • 生成随机文件名，避免覆盖
  • 返回图片访问地址

小图片直接 await file.read() 没问题。更大的文件建议分块写入，视频上传会用到这个写法。

多图片上传

多文件上传可以直接接收 list[UploadFile]。

@app.post("/images/batch")
async def upload_images(files: list[UploadFile] = File(...)):
    results = []

    for file in files:
        suffix = IMAGE_TYPES.get(file.content_type or "")
        if suffix is None:
            results.append({
                "filename": file.filename,
                "ok": False,
                "reason": "unsupported image type",
            })
            continue

        target = MEDIA_DIR / f"{uuid4().hex}{suffix}"
        content = await file.read()
        target.write_bytes(content)

        results.append({
            "filename": target.name,
            "ok": True,
            "url": f"/images/{target.name}",
        })

    return {"files": results}

批量上传不要一遇到坏文件就把全部请求打回去。上面这种返回列表的方式更适合前端展示：哪张成功、哪张失败，一眼能看清。

返回图片：FileResponse

图片保存在本地后，可以用 FileResponse 返回。

import mimetypes

from fastapi.responses import FileResponse


def get_media_path(filename: str) -> Path:
    base = MEDIA_DIR.resolve()
    path = (base / filename).resolve()

    if path != base and base not in path.parents:
        raise HTTPException(status_code=400, detail="非法文件路径")

    if not path.is_file():
        raise HTTPException(status_code=404, detail="文件不存在")

    return path


@app.get("/images/{filename}")
async def read_image(filename: str):
    path = get_media_path(filename)
    media_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    return FileResponse(path, media_type=media_type, filename=path.name)

这里有个小细节：不要直接把用户传进来的 filename 拼到路径里就返回。要做路径归一化，防止 ../ 这类路径穿越。

视频上传：分块写入

视频文件通常比图片大，不适合一次性读进内存。可以边读边写。

from fastapi import Form


VIDEO_TYPES = {
    "video/mp4": ".mp4",
    "video/webm": ".webm",
    "video/quicktime": ".mov",
}


@app.post("/videos")
async def upload_video(
    video: UploadFile = File(...),
    title: str = Form(default=""),
):
    suffix = VIDEO_TYPES.get(video.content_type or "")

    if suffix is None:
        raise HTTPException(status_code=400, detail="只支持 mp4、webm、mov 视频")

    target = MEDIA_DIR / f"{uuid4().hex}{suffix}"

    with target.open("wb") as buffer:
        while chunk := await video.read(1024 * 1024):
            buffer.write(chunk)

    await video.close()

    return {
        "title": title,
        "filename": target.name,
        "content_type": video.content_type,
        "url": f"/videos/{target.name}",
    }

Form 可以和 File 一起用。比如上传视频时，顺手带一个标题、分类、备注，都可以走表单字段。

返回视频：优先用 FileResponse

如果视频已经在磁盘上，FileResponse 是最省心的方式。

FastAPI 文件响应和流式响应

@app.get("/videos/{filename}")
async def read_video(filename: str):
    path = get_media_path(filename)
    media_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    return FileResponse(path, media_type=media_type, filename=path.name)

浏览器访问：

<video controls width="640" src="http://127.0.0.1:8000/videos/demo.mp4"></video>

对本地文件来说，FileResponse 代码少，行为也更符合静态文件返回的直觉。

需要逐块处理时，用 StreamingResponse

如果视频不是一个普通本地文件，比如你要从远端读、边处理边返回、或者分块生成内容，可以用 StreamingResponse。

from collections.abc import Iterator

from fastapi.responses import StreamingResponse


def iter_file(path: Path, chunk_size: int = 1024 * 1024) -> Iterator[bytes]:
    with path.open("rb") as file:
        while data := file.read(chunk_size):
            yield data


@app.get("/videos/{filename}/stream")
def stream_video(filename: str):
    path = get_media_path(filename)
    media_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    return StreamingResponse(iter_file(path), media_type=media_type)

如果只是返回磁盘文件，别急着上流式响应。FileResponse 更简单。StreamingResponse 的价值在于你需要控制内容产生过程。

Python 客户端怎么请求

GET 请求：

import requests


resp = requests.get(
    "http://127.0.0.1:8000/items/100",
    params={
        "q": "keyboard",
        "page": 1,
        "size": 20,
        "available": True,
    },
)

print(resp.status_code)
print(resp.json())

POST JSON：

resp = requests.post(
    "http://127.0.0.1:8000/items",
    json={
        "name": "camera",
        "price": 1299,
        "tags": ["photo", "device"],
    },
)

print(resp.status_code)
print(resp.json())

上传图片：

with open("demo.png", "rb") as file:
    resp = requests.post(
        "http://127.0.0.1:8000/images",
        files={"file": ("demo.png", file, "image/png")},
    )

print(resp.json())

上传视频和表单字段：

with open("demo.mp4", "rb") as file:
    resp = requests.post(
        "http://127.0.0.1:8000/videos",
        data={"title": "demo video"},
        files={"video": ("demo.mp4", file, "video/mp4")},
    )

print(resp.json())

客户端这边要注意：JSON 用 json=，文件上传用 files=，普通表单字段用 data=。这三个参数别混着乱塞。

一份完整版本

把上面的核心代码放到一起，结构大概是这样：

import mimetypes
from collections.abc import Iterator
from pathlib import Path
from uuid import uuid4

from fastapi import FastAPI, File, Form, HTTPException, Query, UploadFile
from fastapi.responses import FileResponse, StreamingResponse
from pydantic import BaseModel, Field

app = FastAPI(title="FastAPI Media Demo")

MEDIA_DIR = Path("media")
MEDIA_DIR.mkdir(exist_ok=True)

IMAGE_TYPES = {
    "image/jpeg": ".jpg",
    "image/png": ".png",
    "image/webp": ".webp",
}

VIDEO_TYPES = {
    "video/mp4": ".mp4",
    "video/webm": ".webm",
    "video/quicktime": ".mov",
}


class ItemIn(BaseModel):
    name: str = Field(min_length=1, max_length=60)
    price: float = Field(gt=0)
    tags: list[str] = []
    description: str | None = None


class ItemOut(ItemIn):
    id: int


STORE: dict[int, ItemOut] = {}


def get_media_path(filename: str) -> Path:
    base = MEDIA_DIR.resolve()
    path = (base / filename).resolve()

    if path != base and base not in path.parents:
        raise HTTPException(status_code=400, detail="非法文件路径")

    if not path.is_file():
        raise HTTPException(status_code=404, detail="文件不存在")

    return path


@app.get("/")
async def index():
    return {"message": "FastAPI is ready"}


@app.get("/items/{item_id}")
async def get_item(
    item_id: int,
    q: str | None = Query(default=None, max_length=50),
    page: int = Query(default=1, ge=1),
    size: int = Query(default=10, ge=1, le=100),
):
    return {
        "item_id": item_id,
        "q": q,
        "page": page,
        "size": size,
    }


@app.post("/items", response_model=ItemOut)
async def create_item(item: ItemIn):
    item_id = len(STORE) + 1
    saved = ItemOut(id=item_id, **item.model_dump())
    STORE[item_id] = saved
    return saved


@app.post("/images")
async def upload_image(file: UploadFile = File(...)):
    suffix = IMAGE_TYPES.get(file.content_type or "")

    if suffix is None:
        raise HTTPException(status_code=400, detail="只支持 jpg、png、webp 图片")

    target = MEDIA_DIR / f"{uuid4().hex}{suffix}"
    content = await file.read()
    target.write_bytes(content)

    return {"filename": target.name, "url": f"/images/{target.name}"}


@app.get("/images/{filename}")
async def read_image(filename: str):
    path = get_media_path(filename)
    media_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    return FileResponse(path, media_type=media_type, filename=path.name)


@app.post("/videos")
async def upload_video(
    video: UploadFile = File(...),
    title: str = Form(default=""),
):
    suffix = VIDEO_TYPES.get(video.content_type or "")

    if suffix is None:
        raise HTTPException(status_code=400, detail="只支持 mp4、webm、mov 视频")

    target = MEDIA_DIR / f"{uuid4().hex}{suffix}"

    with target.open("wb") as buffer:
        while chunk := await video.read(1024 * 1024):
            buffer.write(chunk)

    await video.close()

    return {"title": title, "filename": target.name, "url": f"/videos/{target.name}"}


@app.get("/videos/{filename}")
async def read_video(filename: str):
    path = get_media_path(filename)
    media_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    return FileResponse(path, media_type=media_type, filename=path.name)


def iter_file(path: Path, chunk_size: int = 1024 * 1024) -> Iterator[bytes]:
    with path.open("rb") as file:
        while data := file.read(chunk_size):
            yield data


@app.get("/videos/{filename}/stream")
def stream_video(filename: str):
    path = get_media_path(filename)
    media_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    return StreamingResponse(iter_file(path), media_type=media_type)

这份代码不复杂，但已经覆盖了常见业务接口的骨架。往里加鉴权、数据库、对象存储、日志和限流，就是一个更完整的服务。

写接口时容易踩的坑

GET 请求别塞大块 JSON。查询条件少，用查询参数；条件很复杂，可以考虑 POST 做复杂查询。

文件类型别只看后缀。后缀可以伪造，content_type 也不是绝对可信。严谨场景要读取文件头做二次判断。

不要把上传文件直接原名保存。文件名可能冲突，也可能带奇怪字符。用随机名保存，把原始文件名作为元数据更稳。

不要忽略路径穿越。返回文件时，用户传入的文件名必须限制在媒体目录下。

视频不一定要流式。普通本地文件优先用 FileResponse，需要分块处理再用 StreamingResponse。