相关疑难解决方法(0)

在 FastAPI 中传递字典列表，一般我们会定义一个 pydantic 模式，并会提到

param: List[schema_model]

我面临的问题是我有文件要附加到我的请求中。我找不到在路由器功能中定义架构和文件上传的方法。为此，我将所有参数（请求正文）定义为正文参数，如下所示。

@router.post("/", response_model=DataModelOut)
async def create_policy_details(request:Request,
    countryId: str = Body(...),
    policyDetails: List[dict] = Body(...),
    leaveTypeId: str = Body(...),
    branchIds: List[str] = Body(...),
    cityIds: List[str] = Body(...),
    files: List[UploadFile] = File(None)
    ):

当我使用 postman 的 form-data 选项发送请求时，它为 policyDetails 参数显示“0:value is not a valid dict”。我正在发送 [{"name":"name1","department":"d1"}]。它说不是有效的字典，即使我发送有效的字典。谁可以帮我这个事？DataModelOut 类

class DataModelOut(BaseModel):
    message: str = ""
    id: str = ""
    input_data: dict = None
    result: List[dict] = []
    statusCode: int

我一直在使用FastAPI来创建基于 HTTP 的 API。它目前支持 JSON 编码的参数，但我也想支持form-urlencoded（甚至理想情况下form-data）同一 URL 上的参数。

按照尼基塔的回答，我可以获得单独的网址：

from typing import Optional
from fastapi import FastAPI, Body, Form, Depends
from pydantic import BaseModel

class MyItem(BaseModel):
    id: Optional[int] = None
    txt: str

    @classmethod
    def as_form(cls, id: Optional[int] = Form(None), txt: str = Form(...)) -> 'MyItem':
        return cls(id=id, txt=txt)

app = FastAPI()

@app.post("/form")
async def form_endpoint(item: MyItem = Depends(MyItem.as_form)):
    print("got item =", repr(item))
    return "ok"

@app.post("/json")
async def json_endpoint(item: MyItem = Body(...)):
    print("got …

我正在尝试使用文件上传用户数据。我想做这样的事情，验证用户数据并附加文件

class User(BaseModel):
    user: str
    name: str

@router.post("/upload")
async def create_upload_file(data: User, file: UploadFile = File(...)):
    print(data)
    return {"filename": file.filename}

但它不起作用错误：无法处理的实体响应正文：

{
  "detail": [
    {
      "loc": [
        "body",
        "data"
      ],
      "msg": "value is not a valid dict",
      "type": "type_error.dict"
    }
  ]
}

但如果我将 url 全部工作分开：

class User(BaseModel):
    user: str
    name: str

@router.post("/d")
async def create(file: UploadFile = File(...)):
    return {"filename": file.filename}

@router.post("/")
def main(user: User):
    return user

如何将所有内容结合在一起？

有没有办法将pydantic模型转换为fastapi中的查询参数？

我的一些端点通过主体传递参数，但其他一些端点直接在查询中传递它们。所有这些端点共享相同的数据模型，例如：

class Model(BaseModel):
    x: str
    y: str

我想避免在我的“查询参数端点”的定义中重复我对此模型的定义，例如test_query在此代码中：

class Model(BaseModel):
    x: str
    y: str

@app.post("/test-body")
def test_body(model: Model): pass

@app.post("/test-query-params")
def test_query(x: str, y: str): pass

这样做最干净的方法是什么？

我正在尝试上传图像，但 FastAPI 返回时出现一个我无法弄清楚的错误。

file: UploadFile = File(...)如果我从函数定义中省略“ ”，它就会正常工作。但是当我将 the 添加file到函数定义中时，它会抛出错误。

这是完整的代码。

@router.post('/', response_model=schemas.PostItem, status_code=status.HTTP_201_CREATED)
def create(request: schemas.Item, file: UploadFile = File(...), db: Session = Depends(get_db)):

    new_item = models.Item(
        name=request.name,
        price=request.price,
        user_id=1,
    )
    print(file.filename)
    db.add(new_item)
    db.commit()
    db.refresh(new_item)
    return new_item

Pydantic模型Item只是

class Item(BaseModel):
    name: str
    price: float

错误是：
代码 422 错误：无法处理的实体

{
  "detail": [
    {
      "loc": [
        "body",
        "request",
        "name"
      ],
      "msg": "field required",
      "type": "value_error.missing"
    },
    {
      "loc": [
        "body",
        "request",
        "price"
      ],
      "msg": "field required", …

具体来说，我希望以下示例能够正常工作：

from typing import List
from pydantic import BaseModel
from fastapi import FastAPI, UploadFile, File


app = FastAPI()


class DataConfiguration(BaseModel):
    textColumnNames: List[str]
    idColumn: str


@app.post("/data")
async def data(dataConfiguration: DataConfiguration,
               csvFile: UploadFile = File(...)):
    pass
    # read requested id and text columns from csvFile

如果这不是 POST 请求的正确方法，请告诉我如何从 FastAPI 中上传的 CSV 文件中选择所需的列。

我正在学习使用 FastAPI，并且在实现一个简单的 API 时一遍又一遍地出现此错误，但我无法弄清楚原因

"detail": "There was an error parsing the body"

这发生在我这两个端点上：

完整代码：代码库

片段：

app_v1 = FastAPI(root_path='/v1')

# JWT Token request
@app_v1.post('/token')
async def login_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    jwt_user_dict = {"username": form_data.username, "password": form_data.password}
    jwt_user = JWTUser(**jwt_user_dict)
    user = authenticate_user(jwt_user)
    if user is None:
        return HTTP_401_UNAUTHORIZED
    jwt_token = create_jwt_token(user)
    return {"token": jwt_token}

要求：

@app_v1.post("/user/photo")
async def update_photo(response: Response, profile_photo: bytes = File(...)):
    response.headers['x-file-size'] = str(len(profile_photo))
    response.set_cookie(key='cookie-api', value="test")
    return {"profile photo size": len(profile_photo)}

要求：