小编Man*_*are的帖子

我拥有的

字典：

user_dict = {'user': {'field1': 'value1',
                      'field2': 'value2'},
             'admin':{'field1': 'value3',
                      'field2': 'value4'}}

悬臂模型：

class User(BaseModel):
    account_type: Optional[str] = 'user'
    field1: Optional[str] = ''
    field1: Optional[str] = ''

    class Config:
        validate_assignment = True

    @validator("account_type", pre=True, always=True)
        def _set_account_type(cls, account_type: str):
        return account_type or "user"

期望的结果

我想根据 account_type 从 user_dict 中查找 field1 和 field2 的值。所以我应该能够做这样的事情：

user = User()
print(user.field1)
# > 'value1'

print(user.field2)
# > 'value2'

user2 = User(field1=None, field2=None)
print(user2.field1)
# > 'value1'

print(user2.field2)
# > 'value2'

user3 = User(account_type="admin") …

所以我正在尝试编写几个测试来测试我使用异步 MongoDB 连接的函数。为了连接到 MongoDB，我使用 Motor 和 asyncio。我需要帮助模拟电机连接。

我的代码：

公地.py

mongo = None

黑名单.py

import commons

class Blacklist(object):
    async def check_if_blacklisted(self, word: str):
        blacklisted = False
        if await commons.mongo.dbtest.blacklist.find_one({'word': word}):
            blacklisted = True
        return blacklisted

主文件

import asyncio
from blacklist import Blacklist
from motor.motor_asyncio import AsyncIOMotorClient
import commons

async def run():
    commons.mongo = AsyncIOMotorClient("mongodb://localhost", io_loop=asyncio.get_event_loop())
    blacklist_checker = Blacklist()
    result = await blacklist_checker.check_if_blacklisted(word="should_be_false")
    print(result)
    # > False

    result = await blacklist_checker.check_if_blacklisted(word="should_be_true")
    print(result)
    # > True

loop = asyncio.get_event_loop()
loop.run_until_complete(run())
loop.close()

我现在想通过模拟电机连接来测试 blacklist.py，但我似乎无法正常运行测试。以下是我尝试过的代码： …

如果在字段中传递 None ，我可以在 Pydantic 中设置默认值吗？

我有以下代码，但在我看来，这里的验证器仅适用于模型的初始化，而不适用于其他情况。

我的代码：

class User(BaseModel):
     name: Optional[str] = ''
     password: Optional[str] = ''
     email: EmailStr
    

    @validator('name')
    def set_name(cls, name):
        return name or 'foo'

遇到的问题：

user = User(name=None, password='some_password', email='user@example.com')
print("Name is ", user.name)
# > 'Name is foo'

user.name = None
print("Name is ", user.name)
# > 'Name is None'

期望输出：

user = User(name='some_name', password='some_password', email='user@example.com')
user.name = None
print("Name is ", user.name)
# > 'Name is foo'

关于如何获得所需输出的任何想法？我认为拥有 getter 和 setter 将有助于解决这个问题。但是，我无法让它们在 Pydantic 模型中工作：

尝试实现 …

我正在尝试处理用户上传的文件。但是，我希望用户在上传完成后得到响应并终止连接，但继续处理文件。我的代码看起来像这样：

@app.route("/upload", methods=['POST'])
async def upload(request):
        try:
            file = request.files.get("file")
        except Exception as e:
            return json({"Received": False})

    await process(file)
    return response.text("File has been uploaded successfully")


async def process(file):
    """ Processing File and generate data"""
    ...
    result = await foo(data)
    

在当前框架中，用户必须等待该过程完成。我希望用户得到这样的回复：

“文件已上传成功”

并终止连接但仍继续处理（文件）。