小编ale*_*ame的帖子

我可以从 Pydantic 获取传入的额外字段吗?

我在 Pydantic Config 中定义了 pydantic 架构extra = Extra.allow

是否可以获取单独传递给架构的一个列表或一组额外字段。

例如:

from pydantic import BaseModel as pydanticBaseModel
class BaseModel(pydanticBaseModel):
    name: str

    class Config:
        allow_population_by_field_name = True
        extra = Extra.allow
Run Code Online (Sandbox Code Playgroud)

我传递以下 JSON:

   {
    "name": "Name",
    "address": "bla bla",
    "post": "post"
   }
Run Code Online (Sandbox Code Playgroud)

我需要一个来自 的函数pydantic(如果可用),它将返回传递的所有额外字段。喜欢:{"address", "post"}

python python-3.x pydantic

28
推荐指数
3
解决办法
3万
查看次数

如何从我的 FastAPI 应用程序向另一个站点 (API) 发送 HTTP 请求?

我正在尝试http://httpbin.org/uuid使用以下代码片段一次向服务器发送 100 个请求

from fastapi import FastAPI
from time import sleep
from time import time
import requests
import asyncio

app = FastAPI()

URL= "http://httpbin.org/uuid"


# @app.get("/")
async def main():
    r = requests.get(URL)
    # print(r.text)
    
    return r.text

async def task():
    tasks = [main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main()]
    # print(tasks)
    # input("stop")
    result = await asyncio.gather(*tasks)
    print (result)

@app.get('/')
def f():
    start = time()
    asyncio.run(task())
    print("time: ",time()-start)
Run Code Online (Sandbox Code Playgroud)

我将 FastAPI 与 Asyncio 结合使用,以实现大约 3 秒或更短的最短时间,但使用上述方法我得到的总时间为 66 秒,超过一分钟。我还想保留main用于附加操作的功能r.text。我知道要实现如此短的时间,需要并发性,但我不确定我在这里犯了什么错误。

python httprequest async-await python-asyncio fastapi

19
推荐指数
1
解决办法
7728
查看次数

如何在 FastAPI 中进行多处理

在处理 FastAPI 请求时,我有一个受 CPU 限制的任务要对列表的每个元素执行。我想在多个 CPU 内核上进行此处理。

在 FastAPI 中执行此操作的正确方法是什么?我可以使用标准multiprocessing模块吗?到目前为止,我发现的所有教程/问题仅涵盖 I/O 绑定任务,如 Web 请求。

python multiprocessing python-asyncio fastapi uvicorn

18
推荐指数
1
解决办法
9382
查看次数

Python 3.9 和 PEP 585 中的typing.Any - 标准集合中的类型提示泛型

我想了解是否typing仍然需要该软件包?

如果在 Python 3.8 中我这样做:

from typing import Any, Dict
my_dict = Dict[str, Any]
Run Code Online (Sandbox Code Playgroud)

现在,在通过 PEP 585 的 Python 3.9 中,现在首选使用集合的内置类型,因此:

from typing import Any
my_dict = dict[str, Any]
Run Code Online (Sandbox Code Playgroud)

我是否仍然需要使用typing.Any或者是否有一个我找不到的内置类型来替换它?

python type-hinting python-typing python-3.9

12
推荐指数
1
解决办法
4340
查看次数

FastAPI 依赖项(yield):如何手动调用它们?

FastAPI 使用 Depends() 注入返回或产生的变量。例如,FastAPI/SQL

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
...
def create_user(db: Session = Depends(get_db)):
...
Run Code Online (Sandbox Code Playgroud)

get_db()如果我想在其他地方(FastAPI 路由之外)使用它,我该怎么做?我知道这是Python的核心知识,但我似乎无法弄清楚。我最初的想法是db = yield from get_db(),但我无法调用yield from异步函数(并且不知道它是否还能工作)。然后我尝试:

with get_db() as db:
   pass
Run Code Online (Sandbox Code Playgroud)

由于原始文件get_db()未包装为@contextmanager. (注意,我不想装饰这个 - 我使用get_db作为示例,我需要处理更复杂的依赖关系)。最后,我尝试了db = next(get_db())- 它有效,但我认为这不是正确的解决方案。何时/如何finally调用 - 当我的方法返回时?在其他一些依赖项中,有需要执行的生成后代码;我需要next()再次调用以确保代码执行吗?看来next()方法不太对啊 有任何想法吗?

python starlette fastapi

12
推荐指数
1
解决办法
5252
查看次数

如何为 Pydantic 列表字段提供默认值?

我想创建一个 Pydantic 模型,其中有一个列表字段,未初始化的默认值为空列表。有没有一种惯用的方法来做到这一点?

对于 Python 的内置数据类对象,您可以使用field(default_factory=list),但是在我自己的实验中,这似乎可以防止我的 Pydantic 模型被腌制。一个天真的实现可能是这样的:

from pydantic import BaseModel

class Foo(BaseModel):
    defaulted_list_field: Sequence[str] = [] # Bad!
Run Code Online (Sandbox Code Playgroud)

但是我们都知道不要使用像空列表文字这样的可变值作为默认值。

那么给 Pydantic 列表字段一个默认值的正确方法是什么?

python python-dataclasses pydantic fastapi

11
推荐指数
2
解决办法
4513
查看次数

如何使用 pydantic 创建具有 uuid 的子项

我尝试创建 Foo 的子级,每个子级都应该有自己的 uuid。在真实的代码中,不会创建 Foo 的实例,只会创建它的子实例。子项稍后将保存在数据库中,uuid 用于从数据库中检索正确的对象。

在第一个代码片段中,我尝试使用 init 方法,这会导致 AttributeError。我还尝试使用类方法,这会导致丢失子对象中的所有字段。

如果我的第二个片段每个孩子都会获得相同的 uuid,这对我来说很有意义,因为它作为默认值传递。

我可以将 uuid 创建放入子级中,尽管在使用继承时这感觉不对。

有没有更好的方法为每个孩子创建一个uuid?

# foo_init_.py
class Foo(BaseModel):
    def __init__(self):
          self.id_ = uuid4()
# >>> AttributeError: __fields_set__

# foo_classmethod.py
class Foo(BaseModel):
    @classmethod
    def __init__(cls):
          cls.id_ = uuid4()
# >>> Bar loses id_ fields

Run Code Online (Sandbox Code Playgroud)
from uuid import uuid4, UUID
from pydantic import BaseModel


class Foo(BaseModel):
    id_: UUID = uuid4()


class Bar(Foo):
    pass


class Spam(Foo):
    pass


if __name__ == '__main__':
    b1 = Bar()
    print(b1.id_)  # >>> 73860f46-5606-4912-95d3-4abaa6e1fd2c
    b2 = …
Run Code Online (Sandbox Code Playgroud)

python pydantic fastapi

11
推荐指数
1
解决办法
1万
查看次数

在为函数分配新属性时,如何抑制 mypy 中的“没有属性”错误?

我经常使用以下习惯用法进行静态初始化:

def compute_answer() -> int:
    if compute_answer.ret is None:
        # Do stuff that only happens the first time
        compute_answer.ret = 42
    return compute_answer.ret

compute_answer.ret = None
Run Code Online (Sandbox Code Playgroud)

但是,使用 mypy 进行类型检查会出现以下错误:

compute.py:2: error: "Callable[[], int]" has no attribute "ret"
compute.py:4: error: "Callable[[], int]" has no attribute "ret"
compute.py:5: error: "Callable[[], int]" has no attribute "ret"
compute.py:7: error: "Callable[[], int]" has no attribute "ret"
Run Code Online (Sandbox Code Playgroud)

如何抑制这些错误,尤其是本地错误(例如,仅此函数/属性)?

python typechecking static-initialization mypy python-typing

10
推荐指数
1
解决办法
6630
查看次数

如何在 FastAPI 中保存 UploadFile

我通过 POST 接受该文件。本地保存时,可以使用file.read()读取内容,但是显示通过file.name不正确(16)的名称。当我尝试按此名称查找它时,出现错误。可能是什么问题?

我的代码:

  @router.post(
    path="/po/{id_po}/upload",
    response_model=schema.ContentUploadedResponse,
)
async def upload_file(
        id_po: int,
        background_tasks: BackgroundTasks,
        uploaded_file: UploadFile = File(...)):
    """pass"""
    uploaded_file.file.rollover()
    uploaded_file.file.flush()
    #shutil.copy(uploaded_file.file.name, f'/home/fyzzy/Desktop/api/{uploaded_file.filename}')
    background_tasks.add_task(s3_upload, uploaded_file=fp)
    return schema.ContentUploadedResponse()
Run Code Online (Sandbox Code Playgroud)

python temporary-files python-asyncio fastapi

9
推荐指数
2
解决办法
9251
查看次数

Pydantic 枚举字段未转换为字符串

我试图将类中的一个字段限制为枚举。但是,当我尝试从课堂上获取字典时,它不会转换为字符串。相反,它保留了枚举。我检查了pydantic 文档,但找不到与我的问题相关的任何内容。

这段代码代表了我真正需要的东西。

from enum import Enum
from pydantic import BaseModel

class S(str, Enum):
    am='am'
    pm='pm'

class K(BaseModel):
    k:S
    z:str

a = K(k='am', z='rrrr')
print(a.dict()) # {'k': <S.am: 'am'>, 'z': 'rrrr'}
Run Code Online (Sandbox Code Playgroud)

我试图让 .dict() 方法返回 {'k': 'am', 'z': 'rrrr'}

python pydantic fastapi

8
推荐指数
4
解决办法
5784
查看次数