自定义响应-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")

小贴士

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

FileResponse

异步传输文件作为响应。

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

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

文件响应将包含适当的 Content-LengthLast-ModifiedETag 的响应头。

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 兼容的形式(如dictlist等)

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


相关推荐:

来自系列:fastapi

分类 python下文章:

1.0 爬虫的介绍,和requests模块的简单使用

1.1 数据解析的三种方式。正则表达式, bs4, xpath

2.0 多任务(进程,协程,线程)爬虫:验证码识别,返回头储存,ip代理 介绍。 异步是什么,爬虫异步的方式。线程,进程,介绍

2.0.1 协程的 async/await 实现 爬虫 单线程 + 异步协程的实现

3.0 基于selenium 模块的 爬虫操作。 selenium 是一个用于Web应用程序测试的工具。Selenium测试直接运行在浏览器中,就像真正的用户在操作一样。

更多...

aid 加一试试->

评论([[comments.sum]])

发表

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

发表

2020-11 By Hchuan.

flask & bootstrap-flask

© 2021 HChuan. All rights reserved.

随机占位图来自:fghrsh

互联网ICP备案号:蜀ICP备2020031846号