标签: fastapi

我在 FastAPI 中使用 Pydantic，在 OpenAPI 文档中定义。意识到要根据需要定义值，我需要将值保留为空，如下所示。

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

但是，我想提供带有示例值的 JSON，我可以使用以下语法创建它。

class Item(BaseModel):
    name: str = "my name"
    description: str = "my description"
    price: float = 0.234
    tax: float = 0.20

但这将使它们成为 OpenAPI 文档中不需要的。有没有什么方法可以让我根据需要指定它们，并给出示例值？

我想通过发布请求获取图像，然后阅读它。我正在尝试这样做：

import numpy as np
from PIL import Image
from fastapi import FastAPI, File, UploadFile, HTTPException, Depends
app = FastAPI()
@app.post("/predict_image")
    @logger.catch
    def predict_image(predict_image: UploadFile = File(...)):
        logger.info('predict_image POST request performed')
        try:
            pil_image = np.array(Image.open(predict_image.file))
        except:
            raise HTTPException(
                status_code=HTTP_422_UNPROCESSABLE_ENTITY, detail="Unable to process file"
            )
        pred = pil_image.shape
        logger.info('predict_image POST request performed, shape {}'.format(pred))
    return {'input_shape': pred}

调用post请求返回INFO: 127.0.0.1:59364 - "POST /predict_image HTTP/1.1" 400 Bad Request

如何修复它？

更新：

官方教程中的示例返回相同的异常：

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    return {"filename": file.filename}

我正在使用 FastApi 并且有一个端点。

我有两个长时间运行的函数，我想使用 asyncio 同时运行

因此，我创建了两个函数：

async def get_data_one():
    return 'done_one'

async def get_data_two():
    return 'done_two'

这些函数从外部 Web 服务获取数据。

我想同时执行它们，所以我创建了另一个函数来执行它：

async def get_data():
    loop = asyncio.get_event_loop()
    asyncio.set_event_loop(loop)
    task_1 = loop.create_task(get_data_one)
    task_2 = loop.create_task(get_data_two)
    tasks = (task_1, task_2)
    first, second = loop.run_until_complete(asyncio.gather(*tasks))
    loop.close()

    # I will then perform cpu intensive computation on the results
    # for now - assume i am just concatenating the results
    return first + second

最后，我有我的终点：

@app.post("/result")
async def get_result_async():
    r = await get_data()

    return r …

我有一个在 FastApi 框架中开发的 Api。\n我有一个调用 FastApi 的 Angular 应用程序。\n由于多种原因，如果我可以在FastApi 作为一个简单的 Web 应用程序。这样我就可以作为“localhost”访问 Api，并且 Api 可以是“私有”的。

这可能吗？

谢谢你，\n乔斯\xc3\xa9 克鲁兹

该文档建议引发带有客户端错误的 HTTPException，这很好。但是如何在文档中按照 HTTPException 的模型显示这些特定错误？意思是带有“详细信息”键的字典。

以下不起作用，因为 HTTPException 不是 Pydantic 模型。

@app.get(
    '/test', 
    responses={
        409 : {
            'model' : HTTPException, 
            'description': 'This endpoint always raises an error'
        }
    }
)
def raises_error():
    raise HTTPException(409, detail='Error raised')

我一直收到[Errno 98] Address already in use但是地址没有被使用。我试图更改 ip 和端口，但它没有改变。

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def main():
return {"message": "Helloworld?FastAPI"}


if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)

uvicorn main:app --reload 也试过了，uvicorn main:app --host=172.0.0.2 --port=5000 它让 [Errno 99] error while attempting to bind on address ('172.0.0.2', 5000): cannot assign requested address 我尝试运行一个烧瓶开发服务器，它也在 172.0.0.1 上运行没有问题？

使用 Arch-Manjaro-Linux

我用 nmap 看看有什么大惊小怪的。

但是127.0.0.1 IP上只有 2 个端口在使用

PORT     STATE SERVICE
631/tcp  open  ipp
8000/tcp open  http-alt

我会使用另一个 …

我检查了 PyInstaller 和 FastAPI（超出最大递归深度）并且 Pyinstaller 编译的 Uvicorn 服务器无法正确启动

FastAPI 演示main.py：

import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    return {"hello": "world"}

if __name__ == '__main__':
    uvicorn.run(app, host="0.0.0.0", port=58000, reload=False)

首先运行 pyinstallerpyinstaller -F main.py --clean并添加hidden_imports规范：

hidden_imports=[
                'uvicorn.logging',
                'uvicorn.loops',
                'uvicorn.loops.auto',
                'uvicorn.protocols',
                'uvicorn.protocols.http',
                'uvicorn.protocols.http.auto',
                'uvicorn.protocols.websockets',
                'uvicorn.protocols.websockets.auto',
                'uvicorn.lifespan',
                'uvicorn.lifespan.on',
            ]

它工作得很好，但是当工人大于 1 时，应用程序必须是字符串：

WARNING: You must pass the application as an import string to enable 'reload' or 'workers'.

所以我改为：

if __name__ == '__main__': …

import uvicorn
from fastapi import FastAPI

# 2. Create the app object
app = FastAPI()

# 3. Index route, opens automatically on http://127.0.0.1:8000
class RunModel():
    @app.get('/')
    def index(self):
        return {'message': 'Hello'}

    @app.get('/predict')
    def get_res(self, feat1: float, feat2:float):
        res = feat1 + feat2
        return {'result': f'{res:.4f}'}

run_model = RunModel()
# 5. Run the API with uvicorn
#    Will run on http://127.0.0.1:8000
if __name__ == '__main__':
    uvicorn.run(app, host='127.0.0.1', port=8000)

当我首先运行它时，我收到错误（在终端而不是浏览器中）422 Unprocessable Entity。接下来的事情是，当我访问 http://localhost:8000/docs 时，它似乎期望我输入 3 个/predict路由值，即预期的两个功能和 self …

我有一个 pydantic 类，例如：

from pydantic import BaseModel
class Programmer(BaseModel):
    python_skill: float
    stackoverflow_skill: float
    total_score: float = None

现在我正在根据其他字段计算 total_score：

@validator("total_score", always=True)
def calculat_total_score(cls, v, *, values):
    return values.get("python_skill") + values.get("stackoverflow_skill")

这很好用，但是现在当我更改其中一项技能时：

programmer = Programmer(python_skill=1.0, stackoverflow_skill=9.0)
print(programmer.total_score) # return 10.0
programmer.python_skill=2.0 
print(programmer.total_score) # still return 10.0

我希望 total_score 自动更新。

任何解决方案？TNX！！

我需要为 Pydantic 对象指定一个 JSON 别名。它根本不起作用。

from pydantic import Field
from pydantic.main import BaseModel


class ComplexObject(BaseModel):
    for0: str = Field(None, alias="for")


def create(x: int, y: int):
    print("was here")
    co = ComplexObject(for0=str(x * y))
    return co


co = create(x=1, y=2)
print(co.json(by_alias=True))

这个的输出是{"for" : null而不是{"for" : "2"}

这是真的吗？这样一个简单的用例怎么可能不起作用？