ASa*_*mil 2 python http http-post pydantic fastapi
具体来说,我希望以下示例能够正常工作:
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
Run Code Online (Sandbox Code Playgroud)
如果这不是 POST 请求的正确方法,请告诉我如何从 FastAPI 中上传的 CSV 文件中选择所需的列。
Chr*_*ris 93
根据FastAPI 文档:
\n\n\n您可以
\nForm
在路径操作中声明多个参数,但您\n不能也将Body
您期望接收的字段JSON
声明为,因为\n请求的正文将使用\n 编码,application/x-www-form-urlencoded
而不是application/json
(当表单包含文件时,它是编码为multipart/form-data
)。这不是 FastAPI 的限制,而是协议的一部分
\nHTTP
。
python-multipart
请注意,如果您尚未安装\xe2\x80\x94,则需要先安装\xe2\x80\x94 ,因为上传的文件是作为“表单数据”发送的。例如:
pip install python-multipart\n
Run Code Online (Sandbox Code Playgroud)\n还应该注意的是,在下面的示例中,端点是用 normal 定义的def
,但您也可以使用async def
(根据您的需要)。请查看此答案,def
了解有关 FastAPI 中vs的更多详细信息async def
。
如果您正在寻找如何上传文件和list
字典/JSON数据,请查看这个答案,以及这个答案和这个工作示例的答案(主要基于以下一些方法)。
如此处所述,可以使用File
和同时定义文件和表单字段Form
。下面是一个工作示例。如果您有大量参数并且希望与端点分开定义它们,请查看有关如何创建自定义依赖项类的答案。
应用程序.py
\npip install python-multipart\n
Run Code Online (Sandbox Code Playgroud)\n您可以通过访问下面的模板来测试上面的示例http://127.0.0.1:8000
。如果您的模板不包含任何 Jinja 代码,您也可以返回一个简单的HTMLResponse
.
模板/index.html
\nfrom fastapi import Form, File, UploadFile, Request, FastAPI\nfrom typing import List\nfrom fastapi.responses import HTMLResponse\nfrom fastapi.templating import Jinja2Templates\n\napp = FastAPI()\ntemplates = Jinja2Templates(directory="templates")\n\n\n@app.post("/submit")\ndef submit(\n name: str = Form(...),\n point: float = Form(...),\n is_accepted: bool = Form(...),\n files: List[UploadFile] = File(...),\n):\n return {\n "JSON Payload": {"name": name, "point": point, "is_accepted": is_accepted},\n "Filenames": [file.filename for file in files],\n }\n\n\n@app.get("/", response_class=HTMLResponse)\ndef main(request: Request):\n return templates.TemplateResponse("index.html", {"request": request})\n
Run Code Online (Sandbox Code Playgroud)\n您还可以使用交互式OpenAPI/Swagger UI 自动文档/docs
(例如 )http://127.0.0.1:8000/docs
或使用 Python来测试此示例requests
,如下所示:
测试.py
\n<!DOCTYPE html>\n<html>\n <body>\n <form method="post" action="http://127.0.0.1:8000/submit" enctype="multipart/form-data">\n name : <input type="text" name="name" value="foo"><br>\n point : <input type="text" name="point" value=0.134><br>\n is_accepted : <input type="text" name="is_accepted" value=True><br> \n <label for="file">Choose files to upload</label>\n <input type="file" id="files" name="files" multiple>\n <input type="submit" value="submit">\n </form>\n </body>\n</html>\n
Run Code Online (Sandbox Code Playgroud)\n人们还可以使用 Pydantic 模型以及Dependency来通知/submit
端点(在下面的示例中)参数化变量base
取决于Base
类。请注意,此方法期望base
数据作为query
(而不是 body
)参数,然后对这些数据进行验证并将其转换为 Pydantic 模型(在本例中,即模型Base
)。从 FastAPI 端点返回 Pydantic 模型实例(在本例中为base
)将使用 ,在幕后自动转换为等效的字典/JSON 对象,如本答案jsonable_encoder
中详细解释的。但是,如果您希望自己在端点内完成此操作,则可以使用 Pydantic 的方法,例如,或简单地,如此答案中所述。此外,以下示例期望请求正文中的内容相同。model_dump()
base.model_dump()
dict(base)
data
Files
multipart/form-data
应用程序.py
\nimport requests\n\nurl = \'http://127.0.0.1:8000/submit\'\nfiles = [(\'files\', open(\'test_files/a.txt\', \'rb\')), (\'files\', open(\'test_files/b.txt\', \'rb\'))]\ndata = {"name": "foo", "point": 0.13, "is_accepted": False}\nresp = requests.post(url=url, data=data, files=files) \nprint(resp.json())\n
Run Code Online (Sandbox Code Playgroud)\n同样,您可以使用下面的模板对其进行测试,这一次,该模板使用 JavaScript 修改元素action
的属性form
,以便将form
数据作为query
参数传递到 URL 而不是form-data
.
模板/index.html
\nfrom fastapi import Form, File, UploadFile, Request, FastAPI, Depends\nfrom typing import List\nfrom fastapi.responses import HTMLResponse\nfrom pydantic import BaseModel\nfrom typing import Optional\nfrom fastapi.templating import Jinja2Templates\n\napp = FastAPI()\ntemplates = Jinja2Templates(directory="templates")\n\n\nclass Base(BaseModel):\n name: str\n point: Optional[float] = None\n is_accepted: Optional[bool] = False\n\n\n@app.post("/submit")\ndef submit(base: Base = Depends(), files: List[UploadFile] = File(...)):\n return {\n "JSON Payload": base,\n "Filenames": [file.filename for file in files],\n }\n\n\n@app.get("/", response_class=HTMLResponse)\ndef main(request: Request):\n return templates.TemplateResponse("index.html", {"request": request})\n
Run Code Online (Sandbox Code Playgroud)\n如前所述,要测试 API,您还可以使用 Swagger UI 或 Python requests
,如下面的示例所示。请注意,数据现在应该传递给方法的params
(而不是)data
参数requests.post()
,因为数据现在作为query
参数发送,而不是在请求正文中发送,这是前面方法 1form-data
中的情况。
测试.py
\n<!DOCTYPE html>\n<html>\n <body>\n <form method="post" id="myForm" onclick="transformFormData();" enctype="multipart/form-data">\n name : <input type="text" name="name" value="foo"><br>\n point : <input type="text" name="point" value=0.134><br>\n is_accepted : <input type="text" name="is_accepted" value=True><br> \n <label for="file">Choose files to upload</label>\n <input type="file" id="files" name="files" multiple>\n <input type="submit" value="submit">\n </form>\n <script>\n function transformFormData(){\n var myForm = document.getElementById(\'myForm\');\n var qs = new URLSearchParams(new FormData(myForm)).toString();\n myForm.action = \'http://127.0.0.1:8000/submit?\' + qs;\n }\n </script>\n </body>\n</html>\n
Run Code Online (Sandbox Code Playgroud)\n另一种选择是将正文数据作为Form
JSON 字符串形式的单个参数(类型为 )传递。为此,您需要在服务器端创建依赖函数。
依赖项是“只是一个函数,它可以采用路径操作函数(也称为端点)可以采用的所有相同参数。您可以将其视为没有装饰器的路径操作函数” 。因此,您需要以与端点参数相同的方式声明依赖项(即,依赖项中的参数名称和类型应该是客户端向该端点发送 HTTP 请求时 FastAPI 所期望的名称和类型) , 例如,data: str = Form(...)
)。然后,base
在端点中创建一个新参数(例如,),使用Depends()
依赖函数并将其作为参数传递给它(注意:不要直接调用它,这意味着不要在函数末尾添加括号\ 的名称,而是使用例如 ,Depends(checker)
其中checker
是依赖函数的名称)。每当新请求到达时,FastAPI 将负责调用您的依赖项、获取结果并将该结果分配给端点base
中的参数(例如 )。有关依赖项的更多详细信息,请查看本节中提供的链接。
在这种情况下,应该使用依赖函数来使用data
方法parse_raw
(注意:在 Pydantic V2 中parse_raw
已弃用并替换为model_validate_json
)来解析(JSON 字符串),并data
根据相应的 Pydantic 模型进行验证。如果ValidationError
引发,HTTP_422_UNPROCESSABLE_ENTITY
则应将错误发送回客户端,包括错误消息;否则,该模型的实例(即Base
本例中的模型)将被分配给端点中的参数,可以根据需要使用该参数。示例如下:
应用程序.py
\nimport requests\n\nurl = \'http://127.0.0.1:8000/submit\'\nfiles = [(\'files\', open(\'test_files/a.txt\', \'rb\')), (\'files\', open(\'test_files/b.txt\', \'rb\'))]\nparams = {"name": "foo", "point": 0.13, "is_accepted": False}\nresp = requests.post(url=url, params=params, files=files)\nprint(resp.json())\n
Run Code Online (Sandbox Code Playgroud)\nChecker
依赖类如果您有多个模型并且希望避免checker
为每个模型创建函数,您可以创建一个通用 Checker
依赖项类,如文档中所述(也请参阅此答案以获取更多详细信息),并将其用于每个不同的模型在你的 API 中。例子:
from fastapi import FastAPI, status, Form, UploadFile, File, Depends, Request\nfrom pydantic import BaseModel, ValidationError\nfrom fastapi.exceptions import HTTPException\nfrom fastapi.encoders import jsonable_encoder\nfrom typing import Optional, List\nfrom fastapi.templating import Jinja2Templates\nfrom fastapi.responses import HTMLResponse\n\napp = FastAPI()\ntemplates = Jinja2Templates(directory="templates")\n\n\nclass Base(BaseModel):\n name: str\n point: Optional[float] = None\n is_accepted: Optional[bool] = False\n\n\ndef checker(data: str = Form(...)):\n try:\n return Base.model_validate_json(data)\n except ValidationError as e:\n raise HTTPException(\n detail=jsonable_encoder(e.errors()),\n status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,\n )\n\n\n@app.post("/submit")\ndef submit(base: Base = Depends(checker), files: List[UploadFile] = File(...)):\n return {"JSON Payload": base, "Filenames": [file.filename for file in files]}\n\n\n@app.get("/", response_class=HTMLResponse)\ndef main(request: Request):\n return templates.TemplateResponse("index.html", {"request": request})\n
Run Code Online (Sandbox Code Playgroud)\n如果针对特定 Pydantic 模型验证输入数据对您来说并不重要,但是您希望接收任意JSON 数据并简单地检查客户端是否发送了有效的 JSON 字符串,您可以使用以下:
\n# ... rest of the code is the same as above\n\nclass Other(BaseModel):\n msg: str\n details: Base\n \n \nclass Checker:\n def __init__(self, model: BaseModel):\n self.model = model\n\n def __call__(self, data: str = Form(...)):\n try:\n return self.model.model_validate_json(data)\n except ValidationError as e:\n raise HTTPException(\n detail=jsonable_encoder(e.errors()),\n status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,\n )\n\n\n@app.post("/submit")\ndef submit(base: Base = Depends(Checker(Base)), files: List[UploadFile] = File(...)):\n pass\n\n\n@app.post("/submit_other")\ndef submit_other(other: Other = Depends(Checker(Other)), files: List[UploadFile] = File(...)):\n pass\n
Run Code Online (Sandbox Code Playgroud)\n或者,您可以简单地使用Json
Pydantic 中的类型(如此处所示):
# ...\nfrom json import JSONDecodeError\nimport json\n\ndef checker(data: str = Form(...)):\n try:\n return json.loads(data)\n except JSONDecodeError:\n raise HTTPException(status_code=400, detail=\'Invalid JSON data\')\n\n\n@app.post("/submit")\ndef submit(payload: dict = Depends(checker), files: List[UploadFile] = File(...)):\n pass\n
Run Code Online (Sandbox Code Playgroud)\nrequests
测试.py
\n请注意,在 中JSON
,布尔值使用小写的true
或false
文字表示,而在 Python 中,它们必须大写为True
或False
。
import requests\n\nurl = \'http://127.0.0.1:8000/submit\'\nfiles = [(\'files\', open(\'test_files/a.txt\', \'rb\')), (\'files\', open(\'test_files/b.txt\', \'rb\'))]\ndata = {\'data\': \'{"name": "foo", "point": 0.13, "is_accepted": false}\'}\nresp = requests.post(url=url, data=data, files=files) \nprint(resp.json())\n
Run Code Online (Sandbox Code Playgroud)\n或者,如果您愿意:
\nfrom pydantic import Json\n\n@app.post("/submit")\ndef submit(data: Json = Form(), files: List[UploadFile] = File(...)):\n pass\n
Run Code Online (Sandbox Code Playgroud)\nPS 要使用 Python测试/submit_other
端点(在前面的泛型类中描述) ,请将上例中的属性替换为以下属性:Checker
requests
data
import requests\n\nurl = \'http://127.0.0.1:8000/submit\'\nfiles = [(\'files\', open(\'test_files/a.txt\', \'rb\')), (\'files\', open(\'test_files/b.txt\', \'rb\'))]\ndata = {\'data\': \'{"name": "foo", "point": 0.13, "is_accepted": false}\'}\nresp = requests.post(url=url, data=data, files=files) \nprint(resp.json())\n
Run Code Online (Sandbox Code Playgroud)\n模板/index.html
\nimport requests\nimport json\n\nurl = \'http://127.0.0.1:8000/submit\'\nfiles = [(\'files\', open(\'test_files/a.txt\', \'rb\')), (\'files\', open(\'test_files/b.txt\', \'rb\'))]\ndata = {\'data\': json.dumps({"name": "foo", "point": 0.13, "is_accepted": False})}\nresp = requests.post(url=url, data=data, files=files) \nprint(resp.json())\n
Run Code Online (Sandbox Code Playgroud)\n另一种方法来自github 讨论,并将自定义类与用于将给定字符串转换为 Python 字典的类方法结合JSON
在一起,然后将其用于针对 Pydantic 模型进行验证(请注意,与前面提到的 github 链接,下面的例子使用的是@model_validator(mode=\'before\')
,因为引入了 Pydantic V2)。
与上面的方法 3类似,输入数据应作为字符串Form
形式的单个参数传递JSON
(请注意,data
在下面的示例中使用Body
or定义参数Form
都可以工作,无论 \xe2\x80\x94Form
是直接继承自的类Body
也就是说,FastAPI 仍然期望 JSON 字符串作为form
数据,而不是application/json
,因为在这种情况下,请求将使用multipart/form-data
) 编码正文。因此,上面方法 3中的相同test.py示例和index.html模板也可以用于测试下面的示例。
应用程序.py
\nimport requests\nimport json\n\nurl = \'http://127.0.0.1:8000/submit_other\'\ndata = {\'data\': json.dumps({"msg": "Hi", "details": {"name": "bar", "point": 0.11, "is_accepted": True}})}\n# ... rest of the code is the same as above\n
Run Code Online (Sandbox Code Playgroud)\n另一种解决方案是将文件字节转换为base64
-format 字符串,并将其与您可能想要发送到服务器的其他数据一起添加到 JSON 对象中。我不会强烈推荐使用这种方法;但是,为了完整起见,它已作为进一步的选项添加到此答案中。
我不建议使用它的原因是,使用编码文件base64
本质上会增加文件的大小,从而增加带宽利用率,以及上传文件所需的时间和资源(例如,CPU 使用率) (特别是当多个用户同时使用 API 时),因为需要分别在客户端和服务器端进行 Base64 编码和解码(这种方法仅适用
您不能将表单数据与 json 混合使用。
根据 FastAPI文档:
警告:您可以声明多个
File
和Form
参数的路径运行,但你不能同时申报Body
,您希望收到的JSON字段,请求将身体用编码multipart/form-data
代替application/json
。这不是 FastAPI 的限制,它是 HTTP 协议的一部分。
但是,您可以Form(...)
将额外的字符串附加为form-data
:
from typing import List
from fastapi import FastAPI, UploadFile, File, Form
app = FastAPI()
@app.post("/data")
async def data(textColumnNames: List[str] = Form(...),
idColumn: str = Form(...),
csvFile: UploadFile = File(...)):
pass
Run Code Online (Sandbox Code Playgroud)
我采用了@Chris 的非常优雅的Method3(最初由@M.Winkwns 提出)。不过,我稍微修改了它以适用于任何Pydantic 模型:
from typing import Type, TypeVar
from pydantic import BaseModel, ValidationError
from fastapi import Form
Serialized = TypeVar("Serialized", bound=BaseModel)
def form_json_deserializer(schema: Type[Serialized], data: str = Form(...)) -> Serialized:
"""
Helper to serialize request data not automatically included in an application/json body but
within somewhere else like a form parameter. This makes an assumption that the form parameter with JSON data is called 'data'
:param schema: Pydantic model to serialize into
:param data: raw str data representing the Pydantic model
:raises ValidationError: if there are errors parsing the given 'data' into the given 'schema'
"""
try:
return schema.parse_raw(data)
except ValidationError as e
raise HTTPException(detail=jsonable_encoder(e.errors()), status_code=status.HTTP_422_UNPROCESSABLE_ENTITY)
Run Code Online (Sandbox Code Playgroud)
当您在端点中使用它时,您可以用来functools.partial
绑定特定的 Pydantic 模型:
import functools
from pydantic import BaseModel
from fastapi import Form, File, UploadFile, FastAPI
class OtherStuff(BaseModel):
stuff: str
class Base(BaseModel):
name: str
stuff: OtherStuff
@app.post("/upload")
async def upload(
data: Base = Depends(functools.partial(form_json_deserializer, Base)),
files: Sequence[UploadFile] = File(...)
) -> Base:
return data
Run Code Online (Sandbox Code Playgroud)
归档时间: |
|
查看次数: |
1659 次 |
最近记录: |