跳到内容

JSON 与 Base64 编码的字节数据

如果你的应用程序需要接收和发送 JSON 数据，但其中包含二进制数据，你可以将其编码为 base64。

Base64 对比文件

在将二进制数据编码进 JSON 之前，请先考虑是否可以使用 请求文件 (Request Files) 来上传二进制数据，以及使用 自定义响应 - FileResponse 来发送二进制数据。

JSON 只能包含 UTF-8 编码的字符串，因此它不能直接包含原始字节数据。

Base64 可以将二进制数据编码为字符串，但这样做比原始二进制数据占用更多的字符，因此通常不如直接传输文件高效。

仅当你确实需要在 JSON 中包含二进制数据，且无法使用文件传输时，才使用 base64。

Pydantic bytes

你可以声明一个包含 bytes 字段的 Pydantic 模型，然后在模型配置中使用 val_json_bytes，告诉它在校验输入 JSON 数据时使用 base64。作为校验过程的一部分，它会将 base64 字符串解码为字节数据。

from fastapi import FastAPI
from pydantic import BaseModel


class DataInput(BaseModel):
    description: str
    data: bytes

    model_config = {"val_json_bytes": "base64"}

# Code here omitted 👈

app = FastAPI()


@app.post("/data")
def post_data(body: DataInput):
    content = body.data.decode("utf-8")
    return {"description": body.description, "content": content}

# Code below omitted 👇
👀 完整文件预览 
from fastapi import FastAPI
from pydantic import BaseModel


class DataInput(BaseModel):
    description: str
    data: bytes

    model_config = {"val_json_bytes": "base64"}


class DataOutput(BaseModel):
    description: str
    data: bytes

    model_config = {"ser_json_bytes": "base64"}


class DataInputOutput(BaseModel):
    description: str
    data: bytes

    model_config = {
        "val_json_bytes": "base64",
        "ser_json_bytes": "base64",
    }


app = FastAPI()


@app.post("/data")
def post_data(body: DataInput):
    content = body.data.decode("utf-8")
    return {"description": body.description, "content": content}


@app.get("/data")
def get_data() -> DataOutput:
    data = "hello".encode("utf-8")
    return DataOutput(description="A plumbus", data=data)


@app.post("/data-in-out")
def post_data_in_out(body: DataInputOutput) -> DataInputOutput:
    return body

如果你查看 /docs，它们会显示字段 data 期望接收 base64 编码的字节数据。

你可以发送如下请求：

{
    "description": "Some data",
    "data": "aGVsbG8="
}

提示

aGVsbG8= 是 hello 的 base64 编码。

随后 Pydantic 会解码该 base64 字符串，并在模型中将原始字节数据存入 data 字段。

你将收到如下响应：

{
  "description": "Some data",
  "content": "hello"
}

用于输出数据的 Pydantic bytes

对于输出数据，你也可以在模型配置中使用带有 ser_json_bytes 的 bytes 字段，Pydantic 在生成 JSON 响应时会将字节数据序列化为 base64。

from fastapi import FastAPI
from pydantic import BaseModel

# Code here omitted 👈

class DataOutput(BaseModel):
    description: str
    data: bytes

    model_config = {"ser_json_bytes": "base64"}

# Code here omitted 👈

app = FastAPI()

# Code here omitted 👈

@app.get("/data")
def get_data() -> DataOutput:
    data = "hello".encode("utf-8")
    return DataOutput(description="A plumbus", data=data)

# Code below omitted 👇
👀 完整文件预览 
from fastapi import FastAPI
from pydantic import BaseModel


class DataInput(BaseModel):
    description: str
    data: bytes

    model_config = {"val_json_bytes": "base64"}


class DataOutput(BaseModel):
    description: str
    data: bytes

    model_config = {"ser_json_bytes": "base64"}


class DataInputOutput(BaseModel):
    description: str
    data: bytes

    model_config = {
        "val_json_bytes": "base64",
        "ser_json_bytes": "base64",
    }


app = FastAPI()


@app.post("/data")
def post_data(body: DataInput):
    content = body.data.decode("utf-8")
    return {"description": body.description, "content": content}


@app.get("/data")
def get_data() -> DataOutput:
    data = "hello".encode("utf-8")
    return DataOutput(description="A plumbus", data=data)


@app.post("/data-in-out")
def post_data_in_out(body: DataInputOutput) -> DataInputOutput:
    return body

用于输入和输出数据的 Pydantic bytes

当然，你可以使用同一个配置好的模型，通过 val_json_bytes 处理输入（校验），并通过 ser_json_bytes 处理输出（序列化），从而实现接收和发送 JSON 数据。

from fastapi import FastAPI
from pydantic import BaseModel

# Code here omitted 👈

class DataInputOutput(BaseModel):
    description: str
    data: bytes

    model_config = {
        "val_json_bytes": "base64",
        "ser_json_bytes": "base64",
    }

# Code here omitted 👈

app = FastAPI()

# Code here omitted 👈

@app.post("/data-in-out")
def post_data_in_out(body: DataInputOutput) -> DataInputOutput:
    return body
👀 完整文件预览 
from fastapi import FastAPI
from pydantic import BaseModel


class DataInput(BaseModel):
    description: str
    data: bytes

    model_config = {"val_json_bytes": "base64"}


class DataOutput(BaseModel):
    description: str
    data: bytes

    model_config = {"ser_json_bytes": "base64"}


class DataInputOutput(BaseModel):
    description: str
    data: bytes

    model_config = {
        "val_json_bytes": "base64",
        "ser_json_bytes": "base64",
    }


app = FastAPI()


@app.post("/data")
def post_data(body: DataInput):
    content = body.data.decode("utf-8")
    return {"description": body.description, "content": content}


@app.get("/data")
def get_data() -> DataOutput:
    data = "hello".encode("utf-8")
    return DataOutput(description="A plumbus", data=data)


@app.post("/data-in-out")
def post_data_in_out(body: DataInputOutput) -> DataInputOutput:
    return body