标签: fastapi

有两个类似的 pydantic 对象。唯一的区别是有些字段是可选的。如何只定义一个对象中的字段并扩展到另一个对象中？

class ProjectCreateObject(BaseModel):
    project_id: str
    project_name: str
    project_type: ProjectTypeEnum
    depot: str
    system: str
    ...

class ProjectPatchObject(ProjectCreateObject):
    project_id: str
    project_name: Optional[str]
    project_type: Optional[ProjectTypeEnum]
    depot: Optional[str]
    system: Optional[str]
    ...

我正在使用 FastAPI 来@app.websocket监听传入的 websocket。FastAPI（或下面的 Starlette 或 Uvicorn）如何进行 ping/pong 心跳？这是可配置的吗？我根本无法在文档中找到它。

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/")
def ws(websocket: WebSocket):
    pass

fastapi使用starlette，并且在幕后似乎使用websockets。websockets.connect默认情况下使用 20 秒的 aping_interval和ping_timeout，但我不知道 FastAPI 中是否使用了它。

我想解决的问题：

• 目录的好名字，以便它们的目的很明确。
• 将所有项目文件（包括 virtualenv）保存在一个地方，这样我就可以轻松地复制、移动、存档、删除整个项目或估算磁盘空间使用情况。
• 创建某些选定文件集（例如整个应用程序、存储库或 virtualenv）的多个副本，同时保留我不想克隆的其他文件的单个副本。
• 只需通过 rsyncing 选定的一个目录，即可将正确的文件集部署到服务器。
• 很好地处理前端和后端。

我试图在这个非常基本的 FastAPI 示例中启用 CORS，但它似乎不起作用。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware


app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=['*']
)

@app.get('/')
def read_main():
    return {'message': 'Hello World!'}

这是我得到的回应：

curl -v http://127.0.0.1:8000
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to 127.0.0.1 (127.0.0.1) port 8000 (#0)
> GET / HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/7.64.1
> Accept: */*
>
< HTTP/1.1 200 OK
< date: Fri, 08 Jan 2021 19:27:37 GMT
< server: uvicorn
< content-length: 26
< content-type: application/json
< …

所以我目前正在开发一个为多种 NLP 服务提供服务的 FastAPI 项目。为此，我想提供来自 spacy 和 Huggingface 的不同模型。

由于这些模型非常大，因此为每个发布请求加载模型时的推理时间非常长。我的想法是在 FastAPI 启动时加载所有模型（在 app/main.py 中），但是，我不确定这是否是一个好的选择/想法，或者这种方法是否有一些缺点，因为模型会在缓存中（？）。（信息：我想对项目进行 dockerize，然后将其部署在虚拟机上）

到目前为止，我无法在互联网上找到任何指导，所以我希望在这里得到一个好的答案:)

提前致谢！

在 Fast API 文档中建议使用 .env 来加载配置。只是据我所知它只支持字符串。

from fastapi import FastAPI
from pydantic import BaseSettings


class Settings(BaseSettings):
    api_tokens = []

    class Config:
        env_file = ".env"


settings = Settings()
app = FastAPI()

我通常每隔几个月更改一次 API 令牌，将新的令牌添加到列表中，并在一段时间后删除旧的令牌。这为用户提供了足够的时间升级到最新版本而不会造成任何中断。同时，两个 API 令牌都将在一段时间内有效。

但我无法在文件中定义列表.env。

API_TOKENS = abc123,abc321

我错过了什么吗？

更新：

这实际上是可能的。下面的答案是正确的，但是我仍然必须像这样更改类型：

class Settings(BaseSettings):
    api_tokens: list

我想将一个简单的 FastAPI/uvicorn 部署到 Azure 应用程序服务上。

每次部署时，一切似乎都运行顺利，但当我输入 Web 应用程序的 URL 时，我会收到以下错误消息：

在我的本地机器上，uvicorn 工作正常。在我的网络服务上，我运行 python 3.7 和 fastapi 0.62.0。

每次部署时，我都会启动一个名为startup.sh的文档，其中仅包含一行代码：

python -m uvicorn main:app --host 0.0.0.0 --port 80

非常感谢帮助！

我正在 FastAPI 端点中工作，该端点进行 I/O 绑定操作，该操作是异步的以提高效率。但是，这需要时间，所以我想缓存结果以在一段时间内重复使用。

我目前有这个：

from fastapi import FastAPI
import asyncio

app = FastAPI()

async def _get_expensive_resource(key) -> None:
    await asyncio.sleep(2)
    return True

@app.get('/')
async def get(key):
    return await _get_expensive_resource(key)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("test:app")

我正在尝试使用该cachetools包来缓存结果，并且我尝试了如下操作：

import asyncio
from cachetools import TTLCache
from fastapi import FastAPI
  
app = FastAPI()

async def _get_expensive_resource(key) -> None:
    await asyncio.sleep(2)
    return True

class ResourceCache(TTLCache):
    def __missing__(self, key):
        loop = asyncio.get_event_loop()
        resource = loop.run_until_complete(_get_expensive_resource(key))
        self[key] = resource
        return resource

resource_cache …

我刚刚开始使用 FastAPI、SQLAlchemy、Pydantic，我正在尝试构建一个简单的 API 端点，以将 postgis 表中的行作为 geojson 特征集合返回。

这是我的sqlalchemy模型：

class Poi(Base):
    __tablename__ = 'poi'

    id = Column(Integer, primary_key=True)
    name = Column(Text, nullable=False)
    type_id = Column(Integer)
    geometry = Column(Geometry('POINT', 4326, from_text='ST_GeomFromEWKT'),
                      nullable=False)

使用的geojson_pydantic相关pydantic型号有：

from geojson_pydantic.features import Feature, FeatureCollection
from geojson_pydantic.geometries import Point
from typing import List

class PoiProperties(BaseModel):
    name: str
    type_id: int

class PoiFeature(Feature):
    id: int
    geometry: Point
    properties: PoiProperties

class PoiCollection(FeatureCollection):
    features: List[PoiFeature]

期望的输出：

理想情况下，我希望能够检索并返回数据库记录，如下所示：

def get_pois(db: Session, skip: int = 0, limit: int …

我有一个使用 FastAPI 和 SQLAlchemy 构建的 Web 应用程序，它在本地使用 Docker 运行正常，但在带有托管 Postgres DB 的 DigitalOcean 上，db 查询失败并出现错误：

(psycopg2.OperationalError) 服务器意外关闭了连接\n\t这可能意味着服务器在处理请求之前或处理请求时异常终止\n\t。\n\n（此错误的背景：http ://sqlalche.me/e /14/e3q8 )"}

我之前在使用 Flask 时遇到过这个错误，问题是我必须设置引擎选项pool_pre_ping=True并将我的 cluster/droplet IP 添加到数据库的可信来源。但是看起来对于 FastAPI 这还不够。我还能做些什么来成功执行查询？

背景

• 蟒蛇 3.9
• DigitalOcean 托管 Postgres 13
• psycopg==2.8.6 但也尝试了 2.8.5（100% 在与 Flask 类似的情况下对我有用）和 2.7.4 以防万一
• 我已经pool_pre_ping=True设置
  • 我检查过它确实True在使用请求之前设置为正确session.get_bind().pool._pre_ping，它实际上是True
• 我检查了我的集群节点的 IP 是否在数据库可信来源中
• 我使用一名uvicorn.workers.UvicornH11Worker工人使用 gunicorn 运行该应用程序
• 我使用中间件访问 FastAPI enpoints 中的 db 会话，如下所示：

class DBMiddleware:
    def __init__(self, app, sqlalchemy_uri):
        self.app = app
        self.sqlalchemy_uri = …