我还没有找到该用例的文档。如何获取请求主体，确保它是有效的JSON(任何有效的JSON，包括数字、字符串、布尔值和null，而不仅仅是对象和数组(，并获取实际的JSON。使用Pydantic强制JSON具有特定的结构。
您可以在Request对象中找到几乎所有内容
您可以使用request.json()获取请求主体，这将为您提供解析后的JSON作为字典。
from fastapi import Request, FastAPI
@app.post("/dummypath")
async def get_body(request: Request):
return await request.json()
如果你想以字符串的形式访问正文，你可以使用request.body()
接受的答案也是有效的，但FastAPI提供了一种内置的方法来做到这一点-检查文档中body部分的Singular值。
具有默认Body的参数会获取与传递的Pydantic类型参数不匹配的所有有效载荷(在本例中为整个有效载荷(，并将其转换为适当的Python类型。如果JSON无效，则会产生标准验证错误。
from typing import Any
from fastapi import Body, FastAPI
app = FastAPI()
@app.post('/test')
async def update_item(
payload: Any = Body(None)
):
return payload
UPD:注意第一个Body位置参数(默认值(-这里的None使请求正文可选，...(Ellipsis(-将其标记为必需(不传递任何内容实际上会使其保持必需(。阅读更多关于Ellipsis文档的要求部分
此外，此解决方案适用于仅包含null、true、false、任何字符串、任何数字的JSON。
如果您确信传入数据是"一个有效的JSON">，您可以创建一个简单的类型注释结构来接收任意的JSON数据。
from fastapi import FastAPI
from typing import Any, Dict, AnyStr, List, Union
app = FastAPI()
JSONObject = Dict[AnyStr, Any]
JSONArray = List[Any]
JSONStructure = Union[JSONArray, JSONObject]
@app.post("/")
async def root(arbitrary_json: JSONStructure = None):
return {"received_data": arbitrary_json}示例
1.JSON对象
curl -X POST "http://0.0.0.0:6022/" -H "accept: application/json" -H "Content-Type: application/json" -d "{"test_key":"test_val"}"
响应：
{
"received_data": {
"test_key": "test_val"
}
}
2.JSON数组
curl -X POST "http://0.0.0.0:6022/" -H "accept: application/json" -H "Content-Type: application/json" -d "["foo","bar"]"
响应：
{
"received_data": [
"foo",
"bar"
]
}
如果您不确定传入数据的内容类型，最好解析请求正文。
它可以做为，
from fastapi import FastAPI,Request
app = FastAPI()
@app.post("/")
async def root(request: Request):
return {"received_request_body":await request.body()}这种方法的优点是，主体将包含任何类型的数据，JSON，表单数据，多部分表单数据等。
这是打印Request内容的示例，它将打印json主体(如果它是json可解析的(，否则只打印主体的原始字节。
async def print_request(request):
print(f'request header : {dict(request.headers.items())}' )
print(f'request query params : {dict(request.query_params.items())}')
try :
print(f'request json : {await request.json()}')
except Exception as err:
# could not parse json
print(f'request body : {await request.body()}')
@app.post("/printREQUEST")
async def create_file(request: Request):
try:
await print_request(request)
return {"status": "OK"}
except Exception as err:
logging.error(f'could not print REQUEST: {err}')
return {"status": "ERR"}
FastAPI有一个JSON编码器。
在某些情况下，您可能需要转换数据类型(如Pydantic模型(转换为与JSON 兼容的东西
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
import simplejson as json
class SubmitGeneral(BaseModel):
controllerIPaddress: str
readerIPaddress: str
ntpServer: str
@app.post("/submitGeneral")
async def submitGeneral(data: SubmitGeneral):
data = jsonable_encoder(data)
#data = json.loads(data.json()) # same as above line
print(f"data = {json.dumps(data)}")
# you have to access the properties with brackets, not by dot notation
query = f"update LocalPLC set ControllerIpAddress = '{data['controllerIPaddress']}', ReaderIPAddress = '{data['readerIPaddress']}'"
return {"status": "OK"}
对于那些使用BaseModel并希望拥有JSON字段的用户，可以从pydantic 导入JSON
from fastapi import FastAPI
from pydantic import BaseModel, Json, Field
app = FastAPI()
class MockEndpoint(BaseModel):
endpoint: str = Field(description="API endpoint to mock")
response: Json = Field(description="Example response of the endpoint")
@app.get("/")
async def root():
return {"message": "Hello World"}
@app.post("/mock")
async def mock_request(mock_endpoint: MockEndpoint):
return mock_endpoint
这也是将任何形式的Json作为输入的另一种方法
@app.post("/dict/")
async def post_dict(data: Dict[str, Any]):
return data
但我想这不是最干净的做事方式