6 进阶者使用指南 HCuan

自定义响应-HTML ，流，文件等 ¶
Response cookie 设置 cookie
- 使用 Response 参数
错误处理
对象转为字典 jsonable_encoder
获取路由（ url,url_for
pydantic
- 验证器

自定义响应-HTML ，流，文件等 ¶

默认情况下，FastAPI将使用来返回响应JSONResponse。

你可以通过直接返回 Response 来重载它，参见直接返回响应。

但如果你直接返回 Response，返回数据不会自动转换，也不会自动生成文档（例如，在 HTTP 头 Content-Type 中包含特定的「媒体类型」作为生成的 OpenAPI 的一部分）。

你还可以在 路由操作装饰器 中声明你想用的 Response。

你从 路由操作函数 中返回的内容将被放在该 Response 中。

并且如果该 Response 有一个 JSON 媒体类型（application/json），比如使用 JSONResponse 或者 UJSONResponse 的时候，返回的数据将使用你在路由操作装饰器中声明的任何 Pydantic 的 response_model 自动转换（和过滤）。

介绍的一些响应类型

ORJSONResponse orjson 是一个快速的 Python JSON 库。它基准测试是最快的用于 JSON 序列化的 Python 库，更多
HTMLResponse 之前文章一介绍过，如何使用模板并返回模板生成的网页。在路由函数的装饰器上设置参数@app.get("/items/", response_class=HTMLResponse) 就可以直接返回 HTML 字符串作为网页。。。

其他全部的响应都继承主类 `Response`

Response 类接受如下参数：

content - 一个 str 或者 bytes。
status_code - 一个 int 类型的 HTTP 状态码。
headers - 一个由字符串组成的 dict。
media_type - 一个给出媒体类型的 str，比如 "text/html"

from fastapi import FastAPI, Response

app = FastAPI()


@app.get("/legacy/")
def get_legacy_data():
    data = """<?xml version="1.0"?>
    <shampoo>
    <Header>
        Apply shampoo here.
    </Header>
    <Body>
        You'll have to use soap here.
    </Body>
    </shampoo>
    """
    return Response(content=data, media_type="application/xml")

`RedirectResponse` 重定向

返回 HTTP 重定向。默认情况下使用 307 状态代码（临时重定向）

from fastapi import FastAPI
from fastapi.responses import RedirectResponse

app = FastAPI()

@app.get("/typer")
async def read_typer():
    return RedirectResponse("https://typer.tiangolo.com")

`StreamingResponse` 流式传输迭达器

采用异步生成器或普通生成器/迭代器，然后流式传输响应主体。

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


async def fake_video_streamer():
    for i in range(10):
        yield b"some fake video bytes"


@app.get("/")
async def main():
    return StreamingResponse(fake_video_streamer())

对类似文件的对象使用 `StreamingResponse`

如果您有类似文件的对象（例如，由 open() 返回的对象），则可以在 StreamingResponse 中将其返回。包括许多与云存储，视频处理等交互的库。

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

some_file_path = "large-video-file.mp4"
app = FastAPI()


@app.get("/")
def main():
    file_like = open(some_file_path, mode="rb")
    return StreamingResponse(file_like, media_type="video/mp4")

小贴士

注意在这里，因为我们使用的是不支持 async 和 await 的标准 open()，我们使用普通的 def 声明了路径操作。

`FileResponse`¶

异步传输文件作为响应。

与其他响应类型相比，接受不同的参数集进行实例化：

path - 要流式传输的文件的文件路径。
headers - 任何自定义响应头，传入字典类型。
media_type - 给出媒体类型的字符串。如果未设置，则文件名或路径将用于推断媒体类型。
filename - 如果给出，它将包含在响应的 Content-Disposition 中。

文件响应将包含适当的 Content-Length，Last-Modified 和 ETag 的响应头。

from fastapi import FastAPI
from fastapi.responses import FileResponse

some_file_path = "large-video-file.mp4"
app = FastAPI()


@app.get("/")
async def main():
    return FileResponse(some_file_path)

这里都是部分直接 copy 官方，因为这些方法简单而又重要

使用`Response`参数

您可以Response在路由操作函数中声明该类型的参数。

from fastapi import FastAPI, Response

app = FastAPI()


@app.post("/cookie-and-object/")
def create_cookie(response: Response):
    response.set_cookie(key="fakesession", value="fake-cookie-session-value")
    return {"message": "Come to the dark side, we have cookies"}

然后，向往常一样返回数据就行了

FastAPI将使用该此响应最后的时间来提取 Cookie （还包括标头和状态代码），并将其放入包含您返回的值的最终响应中response_model。

原文

错误处理

要将错误的 HTTP 响应返回给客户端，请使用HTTPException。

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}


@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found",
                            headers={"X-Error": "There goes my error"})

    return {"item": items[item_id]}

# {
#  "detail": "Item not found"
# }

这是一个异常，所以您不能return它，而是使用raise抛出

如果 detail 接收的参数是一个 dict, list ... . 他们将由 Fastapi 主动处理并转换为 json

...

对象转为字典 jsonable_encoder

转换数据类型（如 Pydantic 模型， orm 模型）为与 JSON 兼容的形式（如dict，list等）

jsonable_encoder 在系列数据库迁移远程映射至本地文章也介绍过他，还是非常实用的

from fastapi.encoders import jsonable_encoder


def add_dept(db: Session, item: Pydantic):
    # 假设 item是一个 pydantic对象
    item_dict = jsonable_encoder(item)
    # TbDept是一个 ORM类，将其结构传出类初始化得到 ORM对象
    db_obj = TbDept(**item_dict)
    return db_obj

获取路由（ url,url_for

fastapi 中也是有类似于 flask 的 url_for 的形式的，但在官网并没有看到此介绍。下面内容来自... 的简略说明

对于 fastapi 实例的路由

from fastapi import FastAPI

app = FastAPI()


@app.get('/hello/')
def hello_world():
    return {"msg": "Hello World"}


@app.get('/hello/{number}/')
def hello_world_number(number: int):
    return {"msg": "Hello World Number", "number": number}


print(app.url_path_for('hello_world'))
print(app.url_path_for('hello_world_number', **{"number": 1}))

# ->
# /hello/
# /hello/1/
# /hello/2/

对于 fastapi.apirouter

对于 APIRouter 的实例他并非直接挂载在 fastapi 上，此时应该使用 request.url_for

from fastapi import FastAPI, Request

app = FastAPI()


@app.get('/hello/')
def hello_world():
    return {"msg": "Hello World"}


@app.get('/hello/{number}/')
def hello_world_number(number: int):
    return {"msg": "Hello World Number", "number": number}


@app.get('/')
def named_url_reveres(request: Request):
    return {
        "'hello_world'": request.url_for("hello_world"),
        "'hello_world_number'": request.url_for("hello_world_number", **{"number": 1}),
        "'hello_world_number'": request.url_for("hello_world_number", **{"number": 2})
    }

# Result Response
# {
#     "'hello_world'": "http://0.0.0.0:6022/hello/",
#     "'hello_world_number'": "http://0.0.0.0:6022/hello/1/",
#     "'hello_world_number'": "http://0.0.0.0:6022/hello/2/"
# }

注意传输的第一个参数都是函数名

diss🙃

~~这两种方法仅能获取本实例下的路由地址，对于多个 router 下的路由则不能获取。~~ 可以获取

如：

from fastapi import APIRouter

from apps.api.api_v1.endpoints import img, carousel

api_router = APIRouter()
api_router.include_router(img.router, prefix='/img', tags=['img path'])
api_router.include_router(carousel.router, prefix="/index", tags=["page index"])

# 这两个个实例下的路由是不能互相获取的（第二天：是可以互相获取的，直接传入 路由的函数名就行
# "imgUrl": request.url_for("img.show", **{"file_name": name_list[i]}),
# "imgUrl": request.url_for("img:test")
# ...: router.url_path_for("show")

查文档查了半天，就上面的一个链接实际说明了问题。同时最烦的是源代码没有使用注释。

git hod

而这样一个功能（问题，花了我六个小时，还是没有解决。

第二天，发现可以获取，但是如何路径参数传递"imgUrl": request.url_for("img_show", **{"file_name": name_list[i]}) -> starlette.routing.NoMatchFound

发现原因是 @router.get('/show/{file_name}/', name="展示图片") 路由中的 name 如果指定，则url_for使用路由的函数名无效，必须使用定义的 name。因为自动文档可以用此提示我才会用 name 说明路由作用😂, 所以描述路由还是使用参数 description。

pydantic

验证器

实现对象之间的自定义验证和复杂关系。

# 需要引入一个字典至 pydantic模型，但其中包含了很多无用数据
tset_dict = {
    'title': 'test',
    'item': [
        {   
            'details': {
                'name': 'a'
                'age': 18,
                'phone': '1232121...'
            }
            'xxx': 'xxx',
        },
        # ...
    ]
}

class User(BaseModel):
    name: str,
    age: int

class UserTest(BaseModel):
    title: str,
    details: List[User]  

# 没有验证器初始化模型需要, 深入字典取出数据。（下行代码为实际测试
UserTest(title=test_dict['title'], details=[item['details'] for item in test_dict[item]])

# 加入验证器则
class UserTest(BaseModel):
    title: str,
    details: List[User] = Filed(alias='item')

    @validator('details', each_item=True, pre=True)
    def check_subjects(cls, v):
        return v['details']

UserTest(**test_dict)

@validator():

第一个参数指定要验证的字段名
each_item 适用于List... 等字段，则 v 会接收到传入参数的每一项，否则接收到此字段传入的整个对象
List（此验证器调用顺序大于其他验证器）没有此属性将会导致此验证器后于默认的验证器，而导致报错

部分内容参考自csdn

评论（[[comments.sum]]）

[[item.name]] [[item.email]] [[item.date_time]]

[[itemSon.name]] [[itemSon.email]] [[itemSon.date_time]]

回复 @[[itemSon.reply_user.name]]：

加载更多([[item.son.length - 2]])...

自定义响应-HTML ，流，文件等 ¶

其他全部的响应都继承主类 Response

RedirectResponse 重定向

StreamingResponse 流式传输 迭达器

对类似文件的对象使用 StreamingResponse

FileResponse¶

Response cookie 设置 cookie

使用Response参数