fastapi 请求文件 / 表单 / 处理错误 / 路径操作配置 / jsonable_encoder

简介: fastapi 请求文件 / 表单 / 处理错误 / 路径操作配置 / jsonable_encoder

文章目录


1. File 参数

2. 多文件上传

3. 请求表单与文件

4. 处理错误

5. 自定义响应头

6. 自定义异常处理器

7. 覆盖默认异常处理器

8. 使用 RequestValidationError 的请求体

9. 复用 FastAPI 异常处理器

10. 路径操作参数配置

10.1 status_code,tags

10.2 summary,description

10.3 response description

10.4 deprecated 废除

11. jsonable_encoder() 转换


1. File 参数

from fastapi import FastAPI, Form, File, UploadFile
app = FastAPI()
@app.post("/files/")
async def create_file(file: bytes = File(...)):
    return {"file_size": len(file)}
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    return {"filename": file.filename}


bytes 形式, 文件的所有内容都存储在内存里,适用于小型文件

很多情况下,UploadFile 更好用

1.存储在内存里的文件超出上限,FastAPI 会将其存入磁盘,大型文件不会用尽所有内存

2.可获取上传文件的元数据

3.自带 file-like async 接口

在 async 路径操作函数 内,要用以下方式读取文件内容:


contents = await myfile.read()

在普通 def 路径操作函数 内,则可以直接访问 UploadFile.file

contents = myfile.file.read()

image.png


2. 多文件上传


  • List[bytes], List[UploadFile]
from fastapi import FastAPI, Form, File, UploadFile
from fastapi.responses import HTMLResponse
from typing import List
app = FastAPI()
@app.post("/files/")
async def create_files(files: List[bytes] = File(...)):
    return {"file_sizes": [len(file) for file in files]}
@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile] = File(...)):
    return {"filenames": [file.filename for file in files]}
@app.get("/")
async def main():
    content = """
<body>
<form action="/files/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
<form action="/uploadfiles/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
</body>
    """
    return HTMLResponse(content=content)

image.png

image.png


3. 请求表单与文件


  • FastAPI 支持同时使用 File 和 Form 定义文件和表单字段
@app.post("/f/")
async def create_file(
    file1: bytes = File(...),
    file2: UploadFile = UploadFile(...),
    token: str = Form(...)
):
    return {
        "file_size": len(file1),
        "token" : token,
        "file_content_type" : file2.content_type
        }

image.png


  • 可在一个路径操作中声明多个 FileForm 参数,但不能同时声明要接收 JSON 的 Body 字段。因为此时请求体的编码为 multipart/form-data,不是 application/json

4. 处理错误


  • raise HTTPException()
from fastapi import FastAPI, HTTPException
app = FastAPI()
items = {"foo" : "The Foo items"}
@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!")
    return {"item" : items[item_id]}

触发 HTTPException 时,可以用参数 detail 传递任何能转换为 JSON 的值,不仅限于 str。还支持传递 dict、list 等数据结构

image.png

INFO:     Started server process [12136]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     127.0.0.1:3229 - "GET /items/foo HTTP/1.1" 200 OK
INFO:     127.0.0.1:3240 - "GET /items/bar HTTP/1.1" 404 Not Found
  • 使用 dict 形式的 detail 参数也可以,只要可以被转为 JSON 即可
HTTPException(status_code=404, detail={"msg":"item not found!", "name":["michael","ming"]})

image.png

5. 自定义响应头


  • HTTPException(headers=xxx)
@app.get("/items-header/{item_id}")
async def read_item_header(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]}


image.png


6. 自定义异常处理器


  • 自定义异常类
  • 编写 handler @app.exception_handler(要处理的异常类)
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
class MichaelException(Exception):
    def __init__(self, name: str):
        self.name = name
app = FastAPI()
@app.exception_handler(MichaelException)
async def michael_exception_handler(request: Request, exec: MichaelException):
    return JSONResponse(
        status_code=408,
        content = {"msg": "哦,{}出错了".format(exec.name)})
@app.get("/test/{name}")
async def test(name: str):
    if name == "yoyo":
        raise MichaelException(name)
    return {"test_name" : name}

image.png


7. 覆盖默认异常处理器


from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="3 is not a good number")
    return {"item_id" : item_id}


image.png

  • 更改 RequestValidationError 错误的处理 handler
from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
app = FastAPI()
@app.exception_handler(RequestValidationError)
async def valid_excep_handler(req, exec):
    return PlainTextResponse(str(exec), status_code=400)
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="3 is not a good number")
    return {"item_id" : item_id}

image.png

image.png

  • 同样的,处理 HTTPException 的 handler,自定义处理
from starlette.exceptions import HTTPException as StarletteHTTPException
# 跟 fastapi 的 HTTPException 一样
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(req, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)

or

@app.exception_handler(HTTPException)
async def http_exception_handler(req, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)

image.png


8. 使用 RequestValidationError 的请求体


RequestValidationError 包含其接收到的 无效数据请求的 body 。可以用这个请求体生成日志、调试错误,并返回给用户

from fastapi import FastAPI, Request, status
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel
app = FastAPI()
@app.exception_handler(RequestValidationError)
async def valid_exception_handler(req: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),
    )
class Item(BaseModel):
    title: str
    size: int
@app.post("/items/")
async def create_item(item: Item):
    return item

image.pngimage.png


9. 复用 FastAPI 异常处理器


  • 在自定义处理完异常之后,还可以继续使用 默认的异常处理器
from fastapi import FastAPI, HTTPException
from fastapi.exception_handlers import http_exception_handler, request_validation_exception_handler
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
app = FastAPI()
@app.exception_handler(StarletteHTTPException) # 自定义处理异常
async def custom_http_exception_handler(req, exc):
    print(f"OMG! An HTTP error!: {repr(exc)}")
    return await http_exception_handler(req, exc) # 再调用自带的异常处理器
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(req, exc):
    print(f"OMG! The client sent invalid data!: {exc}")
    return await request_validation_exception_handler(req, exc)
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}
INFO:     127.0.0.1:9594 - "GET /items/abc HTTP/1.1" 422 Unprocessable Entity
OMG! The client sent invalid data!: 1 validation error for Request
path -> item_id
  value is not a valid integer (type=type_error.integer)
INFO:     127.0.0.1:8106 - "GET /items/abc HTTP/1.1" 422 Unprocessable Entity
INFO:     127.0.0.1:10417 - "GET /items/1 HTTP/1.1" 200 OK
OMG! An HTTP error!: HTTPException(status_code=418, detail="Nope! I don't like 3.")
INFO:     127.0.0.1:10417 - "GET /items/3 HTTP/1.1" 418


10. 路径操作参数配置


10.1 status_code,tags


  • tags = [字符串],将反映到 文档中
from typing import Optional, Set
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []
@app.post("/items/", response_model=Item, tags=["items, ming"], status_code=201)
async def create_item(item: Item):
    return item
@app.get("/items/", tags=["items"])
async def read_items():
    return [{"name": "Foo", "price": 42}]
@app.get("/users/", tags=["michael"])
async def read_users():
    return [{"username": "johndoe"}]

image.png


10.2 summary,description


@app.post("/items/", 
            response_model=Item, 
            tags=["items, ming"], 
            status_code=201,
            summary="创建item",
            description="描述item的一些信息"
            )
async def create_item(item: Item):
    return item


image.png


  • description 也可以由 多行注释直接生成,支持 MD 格式
@app.post("/items/", 
            response_model=Item, 
            tags=["items, ming"], 
            status_code=201,
            summary="创建item",
            )
async def create_item(item: Item):
    '''
    多行注释 --> description
    - **name**: each item must have a name
    - **description**: a long description
    - **price**: required
    - **tax**: if the item doesn't have tax, you can omit this
    - **tags**: a set of unique tag strings for this item
    '''
    return item

image.png


10.3 response description

  • response_description 参数
@app.post("/items/", 
            response_model=Item, 
            tags=["items, ming"], 
            status_code=201,
            summary="创建item",
            response_description="响应的描述"
            )
async def create_item(item: Item):
    '''
    多行注释 --> description,支持 MD 格式
    ## 1. 标题
    - **name**: each item must have a name
    - **description**: a long description
    - **price**: required
    - **tax**: if the item doesn't have tax, you can omit this
    - **tags**: a set of unique tag strings for this item
    '''
    return item

16.pngimage.png


10.4 deprecated 废除


  • deprecated
@app.get("/users/", tags=["michael"], deprecated=True)
async def read_users():
    return [{"username": "johndoe"}]

image.png


11. jsonable_encoder() 转换


  • jsonable_encoder()pydantic 模型等数据结构 转换为 与 json 兼容的格式(dict, list 等)
from datetime import datetime
from typing import Optional
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
fake_db = {}
class Item(BaseModel):
    title: str
    timestamp: datetime
    description: Optional[str] = None
app = FastAPI()
@app.put("/items/{id}")
def update_item(id: str, item: Item):
    json_data = jsonable_encoder(item)
    fake_db[id] = json_data
    return fake_db

image.png

这个例子把 Pydantic model 转换为 dict, 把 datetime 转换为 str

相关文章
|
SQL Oracle 关系型数据库
FastAPI数据库系列(一) MySQL数据库操作 一、简介
FastAPI中你可以使用任何关系型数据库,可以通过SQLAlchemy将其轻松的适应于任何的数据库,比如: PostgreSQL MySQL SQLite Oracle Microsoft SQL Server ...
|
7月前
|
Python
Fastapi进阶用法,路径参数,路由分发,查询参数等详解
Fastapi进阶用法,路径参数,路由分发,查询参数等详解
398 1
|
7月前
|
数据采集 前端开发 JavaScript
vue3 + fastapi 实现选择目录所有文件自定义上传到服务器
vue3 + fastapi 实现选择目录所有文件自定义上传到服务器
216 0
FastAPI(54)- 详解 Request 请求对象(上)
FastAPI(54)- 详解 Request 请求对象(上)
621 0
|
7月前
|
IDE 测试技术 开发工具
FastAPI 并发请求解析:提高性能的重要特性
在当今的数字化世界中,网络用户对于高速响应和持续连接的诉求日益显著。这促使了基于 Python 构建的 FastAPI 框架受到广泛关注,它不仅现代化且效率极高,而且简化了并行请求的处理。本篇文章旨在探讨 FastAPI 如何处理这类请求,并对应用实例进行实际编码展示。
【Fastapi】批量上传文件(文档、图片、视频等)
【Fastapi】批量上传文件(文档、图片、视频等)
|
存储 前端开发 JavaScript
聚是一团火散作满天星,前端Vue.js+elementUI结合后端FastAPI实现大文件分片上传
分片上传并不是什么新概念,尤其是大文件传输的处理中经常会被使用,在之前的一篇文章里:[python花式读取大文件(10g/50g/1t)遇到的性能问题(面试向)](https://v3u.cn/a_id_97)我们讨论了如何读写超大型文件,本次再来探讨一下如何上传超大型文件,其实原理都是大同小异,原则就是化整为零,将大文件进行分片处理,切割成若干小文件,随后为每个分片创建一个新的临时文件来保存其内容,待全部分片上传完毕后,后端再按顺序读取所有临时文件的内容,将数据写入新文件中,最后将临时文件再删掉。
聚是一团火散作满天星,前端Vue.js+elementUI结合后端FastAPI实现大文件分片上传
|
存储 SQL 前端开发
FastAPI第三天---文件请求
FastAPI第三天---文件请求
214 0
FastAPI第三天---文件请求
|
前端开发 中间件 测试技术
FastApi的请求拦截
FastApi的请求拦截
883 0
|
前端开发 数据安全/隐私保护
FastApi-12-Form表单
FastApi-12-Form表单
224 0