如何在FastAPI Post请求中添加文件和JSON表单？

Question

具体来说，我希望下面的示例可以工作：

app.py

from fastapi import FastAPI,File,UploadFile,Form,Body
import uvicorn

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}


@app.get("/hello/{name}")
async def say_hello(name: str):
    return {"message": f"Hello {name}"}

@app.post("/jsk")
async def save_image_meta(farm_name: str = Form(...), files: UploadFile = File(...)):
    print('*'*100)
    return {
        "farm_name": farm_name
    }

if __name__ == "__main__":
    uvicorn.run("main:app", host="127.0.0.1", port=8088, reload=True)

如果这不是发布后请求的正确方法，请告知我如何从FastAPI中的上传图像文件中选择所需的列。

test.py

import base64
import hmac
import json
import requests

def parse_params_to_str(params):
    url = "?"
    for key, value in params.items():
        url = url + str(key) + '=' + str(value) + '&'
    return url[1:-1]

def hash_string(qs, secret_key):
    mac = hmac.new(bytes(secret_key, encoding='utf8'), bytes(qs, encoding='utf-8'), digestmod='sha256')
    d = mac.digest()
    validating_secret = str(base64.b64encode(d).decode('utf-8'))
    return validating_secret

def sample_request():
    access_key = "dfaa65a6-ee25-4b03-916b-bedb9095-35f9-4485-a72e-5da4e161a12b"
    secret_key = "GWpnn56bUOnCedhpiUWYJl9dtZ3WllWtUeBUGQva"
    qs = dict(key=access_key)
    header_secret = hash_string(parse_params_to_str(qs), secret_key)
    url = f"http://127.0.0.1:8088/jsk"
    headers = {
        'Content-Type': 'multipart/form-data',
        'Accept': 'application/json',
        'secret': header_secret
    }
    file = "/Users/jskim/Downloads/111111.jpg"
    files = {'files': open(file, 'rb')}
    payload = {"farm_name": "11111"}
    res = requests.post(url, data=payload, headers=headers, files=files)
    return res

sample_request()

我得到400不良请求错误。

如果这不是发布后请求的适当方法，请告知我如何从FastApi中的上传图像文件中选择所需的列。

Answer 1

您不应自己定义content-type多部分/form-data自己。 请求库通过定义边界会自动处理此问题。如果您自己设置此标头，请求将不执行此操作，并且您的服务器将不知道期望什么边界（除非您决定自己也设置边界）。

要工作，您的标题应该没有内容类型：

    headers = {
        'Accept': 'application/json',
        secret: header_secret
    }

Answer 2

我会给你我在我的项目中所做的事情，希望它能帮助您

基本但有帮助

我的模式

class UserProfileSchema(BaseModel):
       first_name: str 
       last_name: str 
       address: Optional[Text]
       image: Optional[str]
       postal_code: Optional[str]
       national_code: Optional[int]

      @classmethod
      def as_form(cls, first_name: str = Form(...), last_name: str = Form(...), address: Optional[Text] = Form(...),
            postal_code: Optional[str] = Form(...),
            national_code: Optional[str] = Form(...)):
           return cls(first_name=first_name,
               last_name=last_name,
               address=address,
               postal_code=postal_code,
               national_code=national_code)
     class Config:
           orm_mode = True

我的视图

    @wrapper_auth('/profile')
    class Profile:
        async def post(profile: UserProfileSchema = Depends(UserProfileSchema.as_form), file: UploadFile = File(...),
                       current_user: User = Security(get_current_user), db: get_session = Depends(get_db)) -> jsonable_encoder:
.......