當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

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

發布時間：2024/7/5 编程问答 28 豆豆

生活随笔收集整理的這篇文章主要介紹了 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() 轉換

learn from https://fastapi.tiangolo.com/zh/tutorial/request-files/

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()

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)

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}

可在一個路徑操作中聲明多個 File 與 Form 參數，但不能同時聲明要接收 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 等數據結構

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"]})

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]}

6. 自定義異常處理器

自定義異常類
編寫 handler @app.exception_handler(要處理的異常類)

from fastapi import FastAPI, Request from fastapi.responses import JSONResponseclass MichaelException(Exception):def __init__(self, name: str):self.name = nameapp = 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}

7. 覆蓋默認異常處理器

from fastapi import FastAPI, HTTPException from fastapi.exceptions import RequestValidationError from fastapi.responses import PlainTextResponseapp = 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}

更改 RequestValidationError 錯誤的處理 handler

from fastapi import FastAPI, HTTPException from fastapi.exceptions import RequestValidationError from fastapi.responses import PlainTextResponseapp = 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}

同樣的，處理 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)

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

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 BaseModelapp = 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: strsize: int@app.post("/items/") async def create_item(item: Item):return item

輸入非整數的 size ，得到報錯 {"detail": exc.errors(), "body": exc.body}

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 StarletteHTTPExceptionapp = 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_idvalue 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, Setfrom fastapi import FastAPI from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Optional[str] = Noneprice: floattax: Optional[float] = Nonetags: 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"}]

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

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

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

10.4 deprecated 廢除

deprecated

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

11. jsonable_encoder() 轉換

jsonable_encoder() 將 pydantic 模型等數據結構轉換為與 json 兼容的格式（dict, list 等）

from datetime import datetime from typing import Optionalfrom fastapi import FastAPI from fastapi.encoders import jsonable_encoder from pydantic import BaseModelfake_db = {}class Item(BaseModel):title: strtimestamp: datetimedescription: Optional[str] = Noneapp = FastAPI()@app.put("/items/{id}") def update_item(id: str, item: Item):json_data = jsonable_encoder(item)fake_db[id] = json_datareturn fake_db

這個例子把 Pydantic model 轉換為 dict, 把 datetime 轉換為 str

總結

以上是生活随笔為你收集整理的fastapi 请求文件 / 表单 / 处理错误 / 路径操作配置 / jsonable_encoder的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： LeetCode 474. Ones a
下一篇： LeetCode 1570. 两个稀疏向