Poetry 使用文档
Poetry 简介
Poetry 是一个用于管理 Python 依赖项的工具。它使用一个名为 pyproject.toml 的文件来指定项目的依赖项，并提供了一个命令行工具来安装和管理这些依赖项。
Poetry 的优势
依赖锁定：Poetry通过一个poetry.lock 的文件将项目的依赖项锁定到一个特定的版本，这可以确保安装的依赖版本完全一致。
依赖变更：Poetry依赖版本变更时，会同步更新pyproject.toml，无需手动处理requirements.txt
依赖包管理：Poetry 可以定义不同环境的依赖，不需要再维护requirements-dev.txt、requirements-test.txt、requirements-pro.txt等多个文件。
-
残留依赖清理：当使用pip uninstall卸载某个依赖时，其相关依赖并不会被清理，导致占用额外空间，如图
[站外图片上传中...(image-1ab1cf-1751965482239)]
如果使用poetry add kac-api==0.2.0安装后，通过poetry remove kac-api卸载，则可以将其关联的依赖卸载干净，这可以有效解决版本一致，但实际包内容变更的场景
[站外图片上传中...(image-baa94e-1751965482239)]
Poetry 的安装
Poetry 可以通过以下命令安装,该命令是在虚拟环境外执行：
pip install poetry
Poetry 的使用
poetry init # 初始化，与poetry new类似，但poetry new会创建项目文件，用于从头开始创建整个项目
在一个项目中首次使用，通过上面命令，会在项目根目录下创建一个名为 pyproject.toml 的文件，并在其中指定项目的依赖项，如下
[tool.poetry]
name = "meta"
version = "0.1.0"
description = ""
authors = ["Your Name <your@email.com>"]
[tool.poetry.dependencies] #生产环境的依赖列表
python = "^3.6.2"
[tool.poetry.dev-dependencies] #测试环境的依赖列表
pytest = "^6.2"
Poetry 还可以用于管理项目的虚拟环境：
poetry shell # 激活项目的虚拟环境(环境不存在，自动创建)
poetry env list 查看poetry管理下的虚拟环境列表
poetry env remove venv 删除虚拟环境
然后可以通过以下命令安装项目的依赖项：
poetry install
- --remove-untracked 移除lockfile之外的依赖，比如通过pip install安装的，高版本使用--sync
如果想要增加新的依赖可以使用poetry add或者直接修改pyproject.toml,然后执行poetry install
poetry add kac-api
当依赖发生变更后，需要更新lock文件(没有则新建)，lock文件可以确保在不同环境中安装相同的依赖项版本，以提供项目的可重现性和一致性，因为poetry install时，项目会根据lock文件中的依赖版本去安装，而不是根据pyproject.toml设置的范围动态选择
poetry lock
poetry show 可以用查看所有依赖，与 pip list类似
(.venv) amiter@amiterdeMacBook-Pro kmc % poetry show | grep kac
kac-api 0.2.0 canway kingeye kac_api
(.venv) amiter@amiterdeMacBook-Pro kmc %
有时候可能会遇到依赖变更，但版本没变的情况，就需要清理掉缓存,否则会提示依赖冲突的报错；这种情况本身是不合理的，应该避免覆盖版本的情况，应该遵循新的SDK发布方案
poetry cache clear pypi:kac-api:0.2.0
删除某个依赖，及其本身的依赖
poetry remove
如果是已经存在的项目且不想额外创建虚拟环境时，需要图中配置设置为true，poetry会优先在项目根目录下使用虚拟环境
[站外图片上传中...(image-2418a5-1751965482239)]
Poetry 的更多信息
https://python-poetry.org/docs/
Pytest 使用文档
简单示例
编写测试用例
在项目中创建一个名为tests的文件夹，并在其中创建一个名为test_demo.py的Python文件
def add(x, y):
return x + y
def test_add():
assert add(2, 3) == 5
断言
断言是测试过程中用于验证预期结果的关键部分，用于比较值、集合、异常等，如下在测试用例中使用断言方法的示例：
def test_add():
assert add(2, 3) == 5
def test_list_contains_element():
assert 2 in [1, 2, 3]
def test_monitor_source_strategy_1(data):
# 测试数据校验
with pytest.raises(ValidationError):
SaveStrategy(**data).dict()
运行测试
在项目根目录下，运行以下命令：
pytest tests
- -s 可以打印出测试函数的print输出
- -v 显示每个测试函数的执行是否成功
- -x 遇到执行失败立即退出
Pytest将自动发现并执行测试用例，并提供相应的输出和结果。
pytest配置方式
1.最常见的就是pytest.ini
[pytest]
testpaths = tests # 用例搜索路径
python_files = # 还可以指定类或函数级别
tests.py
test_*.py
addopts = --strict-markers # 添加一些通用的额外参数
markers = # 自定义标记，执行pytest -m slow 只执行带有此标记的用例
slow: marks tests as slow (deselect with '-m \"not slow\"')
2.结合poetry使用,直接在pyproject.toml中定义配置即可
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["tests.py", "test_*.py"]
addopts = "--strict-markers"
markers = ["slow: marks tests as slow (deselect with '-m \"not slow\"')"]
...
使用场景
钩子函数
setup、teardown 分别在每条测试用例执行前和结束后执行，可以做一些数据准备，以及清理的操作
setup_class、teardown_class分别在每个测试类初始化前后执行，可以定义一些当前类中的测试方法共享的内容，而不需要每个用例都去创建
class TestDemo:
def test_demo(self):
print("test_demo 执行")
assert True
def test_demo1(self):
print("test_demo1 执行")
assert True
def setup_class(cls):
print("setup_class 执行")
def teardown_class(cls):
print("teardown_class 执行")
def setup(self):
print("setup 执行")
def teardown(self):
print("teardown 执行")
============================= test session starts ==============================
collecting ... collected 2 items
test_demo.py::TestDemo::test_demo
test_demo.py::TestDemo::test_demo1
============================== 2 passed in 6.66s ===============================
Process finished with exit code 0
setup_class 执行
setup 执行
Using existing test database for alias 'default' ('test_kmc_saas')...
PASSED [ 50%]test_demo 执行 None
teardown 执行
setup 执行
PASSED [100%]test_demo1 执行 None
teardown 执行
teardown_class 执行
fixture使用
1.当一个函数或方法被fixture装饰后，就可以被当做测试用例或者其他fixtures的入参来使用，pytest会根据入参的名称去查找对应的fixture，找到就会执行并把执行的结果返回，如图bake_node会创建一个拨测节点实例，在函数中可以继续引用
class TestGetUptimeCheckNodeList:
@pytest.fixture
def bake_node(self):
return baker.make(UptimeCheckNode, bk_biz_id=2, is_deleted=False, ip="10.11.25.115", node_id=10001)
@pytest.mark.parametrize(...
def test_select_uptime_check_node(
self,
mocker,
esb_bk_monitor_baseurl: ft.esb_bk_monitor_baseurl,
admin_client,
request_params,
hosts_msg,
return_value,
bake_node,
):
...
2.配合yield实现测试用例的前后执行(类似钩子函数)
如图yield前为生成fixture时执行的内容，之后则为销毁的内容
@fixture(autouse=True)
def clear_caches(db, redis_caches):
yield
# 清理缓存
for cache in redis_caches.all():
cache.clear()
3.使用autouse参数，可以使其自动执行，而无需引入,但这仅在当前module下生效，如果其他module想用，仍需要导入
@pytest.fixture(autouse=True)
def ss():
print("aaaaa")
class TestDemo001:
@pytest.mark.parametrize("a", [1, 2, pytest.param(3, marks=pytest.mark.skip)])
def test_demo(self, a):
print("test_demo 执行", a)
assert True
@pytest.mark.parametrize("a", [1, 2, pytest.param(3, marks=pytest.mark.skip)])
def test_demo1(self, a):
print("test_demo1 执行", a)
assert True
fixture作用域
- function（默认值）：在测试函数被调用时创建，调用完成后销毁
- class：在class中的最后一个测试函数调用完成后，fixture销毁
- module：在module中最后一个测试函数调用完成后，fixture销毁
- package：在package的最后一个测试函数调用完成后，fixture销毁
- session：整个测试会话运行一次，fixture在测试session结束时被销毁
内置fixture
通过pytest --fixtures 查看所有的fixture
db ：可以用于执行数据库的操作
transactional_db：可以开启事务，相当于TransactionTestCase
admin_client : 模拟管理员的client
-
settings ：用于访问和修改 pytest 的配置选项
@fixture(autouse=True) def celery_always_eager_enabled(settings): """开启celery的eager模式""" settings.CELERY_ALWAYS_EAGER = True yield -
cache ：与django的cache不是同一个,可以用来缓存和共享与测试相关数据，加快执行速度
[站外图片上传中...(image-58887f-1751965482239)]
当我们通过命令行执行测试后（通过pycahrm执行不会产生），就会在当前目录下生成一个.pytest_cache文件夹，该文件夹中就保存了测试相关的数据以及缓存的数据，通过pytest -cache-show也可以查看
(.venv) amiter@amiterdeMacBook-Pro kmc % pytest --cache-show ================================================================================ test session starts ================================================================================ platform darwin -- Python 3.6.8, pytest-7.0.1, pluggy-1.0.0 benchmark: 3.4.1 (defaults: timer=time.perf_counter disable_gc=False min_rounds=5 min_time=0.000005 max_time=1.0 calibration_precision=10 warmup=False warmup_iterations=100000) django: settings: settings (from option) rootdir: /Users/amiter/PycharmProjects/kmc, configfile: pyproject.toml, testpaths: tests plugins: benchmark-3.4.1, Faker-14.2.1, lambda-1.2.6, mock-3.6.1, testmon-1.3.3, requests-mock-1.11.0, cov-4.0.0, django-4.5.2, lazy-fixture-0.6.3 cachedir: /Users/amiter/PycharmProjects/kmc/.pytest_cache ------------------------------------------------------------------------------- cache values for '*' -------------------------------------------------------------------------------- cache/stepwise contains: [] name contains: 'jack' test_cache contains: 2pytest --cache-clear 执行测试前先清空该缓存文件
如下是一个全局缓存的示例
# Conftest.py @fixture(autouse=True, scope="session") def init_test_cache(request): request.config.cache.set("test_cache", 0) print(request.config.cache.get("test_cache", None), "test cache start") yield print(request.config.cache.get("test_cache", None), "test cache end")# test_demo.py def test_addition(cache): print(cache.get("test_cache", None), "test_addition cache") assert 2 + 2 == 4 django: settings: settings (from option) rootdir: /Users/amiter/PycharmProjects/kmc, configfile: pyproject.toml collected 1 item tests/test_demo.py::test_addition 0 test cache start 0 test_addition cache PASSED0 test cache end =============== 1 passed in 2.91s =============================================如上图可以看到我们插入的test_cache取到了值，这里的缓存在下次执行测试时仍可使用，可以将一些计算耗时的操作进行缓存，提升测试执行速度
conftest文件
conftest.py文件是固定写法，可以共享本地固定fixture和钩子函数，一般存在于测试目录的顶层,主要作用是可以避免导入，定义在该文件的fixture都可以直接在用例中引入
@fixture(autouse=True, scope="session")
def exec_time():
start_time = time.time()
yield
print(f"本次此时执行时间：{time.time()-start_time}")
跳过测试
有时候需要跳过某些测试，例如当测试依赖于特定的环境或条件时，pytest提供了多种方法来实现测试的跳过。
跳过测试的示例：
import pytest
@pytest.mark.skip(reason="跳过原因")
def test_function_to_skip():
assert 1 == 2
@pytest.mark.skipif(1==2, reason="跳过原因")
def test_function_to_skip_if_condition():
assert 1 == 2
也可以用pytest -m 去指定某些标记，去执行，例如 pytest -m "slow"
@pytest.mark.slow
def test_slow():
pass
参数化测试
@pytest.mark.parametrize
- 使用不同的输入参数在单个测试函数中多次运行相同的测试代码
class TestGetUptimeCheckTaskList:
@classmethod
@pytest.fixture
def create_task(cls):
"""数据库插入两条数据"""
return [
baker.make(
UptimeCheckTask, task_id=i["id"], protocol=i["protocol"], config=i["config"], bk_biz_id=i["bk_biz_id"]
)
for i in cls.return_value["data"]
]
@pytest.mark.parametrize(
"query_params, return_value, result",
[
({"url": "http://www.baidu.com"}, return_value, [10001]), # 存在的url
({"url": "http://www.xxxxx.com"}, return_value, []), # 不存在的url
({}, return_value, [10001, 10002]), # 不传参数，全部返回
],
)
def test_format_task_params(
self,
backend_esb_client: ft.backend_esb_client,
esb_bk_monitor_baseurl: ft.esb_bk_monitor_baseurl,
faker: ft.faker,
query_params,
return_value,
result,
create_task,
mocker,
):
mocker.patch(
"home_application.views.monitor_scene.uptime_task_mgmt.search_uptime_check_task_list",
return_value=(2, return_value["data"]),
)
# 准备数据
viewset = UptimeCheckTaskViewSet()
class MockRequest(object):
...
viewset.request = MockRequest()
viewset.request.query_params = query_params
task_alarm_map = {
"10001_total": 10001,
"10001__alarm_level": "remind",
"10002_total": 10002,
"10002__alarm_level": "remind",
}
queryset_tuple = [(task, "蓝鲸", task_alarm_map) for task in UptimeCheckTask.objects.values()]
ret_data = MultiArgsThreadPool().map(viewset.format_task_params, queryset_tuple)
ret_data = list(filter(lambda x: x, ret_data))
assert len(ret_data) == len(result)
assert [i["task_id"] for i in ret_data] == result
在上面的例子中，test_format_task_params函数将使用不同的参数组合进行三次运行。
@pytest.mark.parametrize可以用于装饰整个测试类，并将所有所有参数传入-
@pytest.mark.parametrize可以多次装饰同一个测试方法，测试方法将会根据参数个数做笛卡尔积生成不同的参数组合，下面的测试函数将会执行9次import pytest @pytest.mark.parametrize("bk_biz_id", [1, 2, 3]) @pytest.mark.parametrize("task_name", ["tcp001", "tcp002", "tcp003"]) def test_get_task_list(bk_biz_id, task_name): ... -
@pytest.mark.parametrize可以对某个具体的参数进行标记
import pytest @pytest.mark.parametrize("task_name,["tcp001",pytest.param("tcp002",marks=pytest.mark.skip)]) def test_search_task_detail(bk_biz_id, task_name): ... ============================= test session starts ============================== collecting ... collected 3 items test_demo.py::TestDemo001::test_demo[1] test_demo.py::TestDemo001::test_demo[2] test_demo.py::TestDemo001::test_demo[3] ========================= 2 passed, 1 skipped in 6.81s ========================= ... Using existing test database for alias 'default' ('test_kmc_saas')... PASSED [ 33%]test_demo 执行 1 PASSED [ 66%]test_demo 执行 2 SKIPPED (unconditional skip) [100%] Skipped: unconditional skip 本次执行时间：6.623747110366821
makers使用
makers用于对测试用例进行分类和控制测试的行为,可以使用 pytest --markers 查看所有的标记
支持多选，假设按测试用例的重要性做了分级， pytest tests -vs -m "p0 and p1"表示被这两个标记的
-
常用的内置标记
-
skip：跳过测试用例，不执行它。 -
xfail：标记测试用例为预期失败，即如果测试用例失败，不会报告为错误。 -
parametrize：允许为测试用例参数化，以便多次运行同一个测试用例的不同参数组合。 -
fixture：指定一个测试用例所需的固定的测试环境或预置条件。 -
usefixtures：指定一个测试用例使用指定的fixture。 -
pylint：标记测试用例为需要进行静态代码分析（例如使用Pylint）的测试。 -
timeout：为测试用例设置一个超时时间，如果测试用例执行时间超过指定的时间，将会被中断。 -
repeat：指定一个测试用例需要被重复执行的次数。
-
自定义标记示例
-
slow ：可以直接在pyproject.toml中定义makers
[tool.pytest.ini_options] testpaths = ["tests"] python_files = ["tests.py", "test_*.py"] addopts = "--strict-markers" markers = ["slow: marks tests as slow (deselect with '-m \"not slow\"')"]
如下，当使用pytest tests -vs -m "slow"时，pytest只执行了被标记的test_demo方法，如果需要执行没被标记的,只需pytest tests -vs -m "not slow"
class TestDemo:
@pytest.mark.slow #使用了slow标记
def test_demo(self):
print("test_demo 执行")
assert True
def test_demo1(self):
print("test_demo1 执行")
assert True
(.venv) amiter@amiterdeMacBook-Pro kmc % pytest tests -vs -m slow
========================================================================= test session starts =========================================================================
platform darwin -- Python 3.6.8, pytest-7.0.1, pluggy-1.0.0 -- /Users/amiter/PycharmProjects/kmc/.venv/bin/python
cachedir: .pytest_cache
rootdir: /Users/amiter/PycharmProjects/kmc, configfile: pyproject.toml, testpaths: tests
collected 2 items / 1 deselected / 1 selected
tests/test_demo.py::TestDemo
test_demo 执行
PASSED
mocker使用
mocker是pytest-mock插件提供的，基于unittest中mock模块，用于模拟单测中的一些方法的行为和返回内容，且语法更简单，无需显示创建Mock实例
-
模拟函数的返回值
def test_search_event_by_page( self, backend_esb_client: ft.backend_esb_client, esb_bk_monitor_baseurl: ft.esb_bk_monitor_baseurl, faker: ft.faker, alarm_event, mocker, ): # mock search_event的返回值 mocker.patch( "home_application.tasks.search_event", return_value={ "16968544551223039": {"status": "CLOSED", "end_time": "2023-10-09 21:06:53"}, }, ) # mock 异步方法 mocker.patch("home_application.tasks.push_event_to_kac") search_event_by_page([(alarm_event.bk_biz_id, alarm_event.alarm_id, alarm_event.begin_time)]) assert AlarmEvent.objects.filter(alarm_id=alarm_event.alarm_id).exists() -
模拟函数的行为
class MyClass: def get_value(self): return 42 def test_get_value(mocker): my_obj = MyClass() mocker.patch.object(my_obj, "get_value").return_value = 100 result = my_obj.get_value() assert result == 100@pytest.mark.parametrize( "biz_strategy_data_map,strategy_item_id,return_value", [ (biz_strategy_data_map_system, 1, [return_value1]), (biz_strategy_data_map_system, 1, [return_value2, return_value1]), ], ) def test_create_or_update(self, biz_strategy_data_map, strategy_item_id, return_value, mocker): mocker.patch( "kmc.monitor_template.service.distribution_strategy_service.DistributionStrategyService.client" ".monitor_v3.save_alarm_strategy_v2", side_effect=return_value)这里要注意当return_value 和 side_effect同时定义，side_effect生效
注意点
1.以类的方式编写测试用例时，类中不能定义__init__方法，否则pytest无法搜集到该类下的用例
2.开启数据库访问 pytestmark = pytest.mark.django_db
3.mocker的导入路径，当要mock的对象是从其他模块导入的时候，patch中的路径要声明为当前模块，例如b模块中引用了from a import search_event
mocker.patch("home_application.a.search_event"） ❌
mocker.patch("home_application.b.search_event"） ✅
pydantic使用
pydantic优点
-
数据解析和序列化
将原始数据（如 JSON、字典等）转换为 Pydantic 模型的实例。同时，它还支持将 Pydantic 模型转换为 JSON、字典等其他格式的数据，方便数据的序列化和反序列化
-
声明式数据校验
通过类型注解和校验器定义字段类型，自动校验和转换数据，确保数据类型及格式符合预期，避免因参数类型导致bug
-
性能占优
def pydantic_performance_test(): model = CreateUptimeCheckTask(**data) return model.dict() def drf_serializer_performance_test(): serializer = UptimeCheckTaskSerializer(data=data) serializer.is_valid() return serializer.validated_data pydantic_time = timeit.timeit(pydantic_performance_test, number=10000) drf_serializer_time = timeit.timeit(drf_serializer_performance_test, number=10000) print(f"Pydantic performance: {pydantic_time} seconds") print(f"Drf Serializer performance: {drf_serializer_time} seconds") Pydantic performance: 0.6493919999338686 seconds Drf Serializer performance: 10.693156832945533 seconds 文档生成
快速使用
模型定义
1.pydantic中定义对象的主要方法是通过继承类BaseModel
from pydantic import BaseModel,Field
from typing import List
class SwitchStrategy(BaseModel):
ids: List[int] = Field(description="策略ID列表")
is_enabled: bool = Field(description="是否启用")
SwitchStrategy(ids=[1,"2"],is_enabled=True).dict() # StrictInt
Out[20]: {'ids': [1, 2], 'is_enabled': True}
pydantic 默认会对传入数据的类型做转换，上面SwitchStrategy的ids我传了字符串，最后输出的是int，如果不想要强制转换，pydantic还提供了严格类型，StrictInt、StrictStr、StrictBytes、StrictInt、StrictFloat、StrictBool
-
dict() : 将一个模型实例转换为字典
- exclude_unset 默认 False
- exclude_none 默认 False
- exclude_defaults 默认 False
- exclude 返回字典中包含的字段
- include 返回字典中不包含的字段
- by_alias 是否将字段别名用作返回字典中的键
json() : 将一个模型实例转换为json，参数与上面一致
-
copy() :复制模型
-
deep 是否深拷贝
# 默认shallow inst = SwitchStrategy(ids=[1,"2"],is_enabled=True) inst_copy_shallow = inst.copy() inst.ids Out[100]: [1, 2] inst.ids.append(3) inst.ids Out[102]: [1, 2, 3] inst_copy_shallow.ids Out[103]: [1, 2, 3] # deep inst = SwitchStrategy(ids=[1,"2"],is_enabled=True) inst_copy_deep = inst.copy(deep=True) inst.ids Out[105]: [1, 2] inst.ids.append(4) inst.ids Out[107]: [1, 2, 4] inst_copy_deep.ids Out[108]: [1, 2] -
update 更新复制后的字典
SwitchStrategy(ids=[1,"2"],is_enabled=True).copy(update={"ids":[3,4]}).dict() Out[82]: {'ids': [3, 4], 'is_enabled': True}
-
-
parse_obj():跟SwitchStrategy(**{"ids":[1,2,3]})类似，对参数要求为dict，keyword参数会报错
class Plugin(BaseModel): os_type: str = Field(default="") plugin_id: str = Field(default="") plugin = Plugin() p_dict = plugin.dict() p_json = plugin.json() plugin1 = Plugin.parse_obj(p_dict) plugin == plugin1 Out[75]: True plugin2 = Plugin.parse_raw(p_json) plugin == plugin2 Out[77]: True plugin1 = Plugin.parse_obj(**p_dict) Traceback (most recent call last): File "/Users/amiter/PycharmProjects/kmc/.venv/lib/python3.6/site-packages/IPython/core/interactiveshell.py", line 3343, in run_code exec(code_obj, self.user_global_ns, self.user_ns) File "<ipython-input-79-474933518c21>", line 1, in <module> plugin1 = Plugin.parse_obj(**p_dict) File "pydantic/main.py", line 513, in pydantic.main.BaseModel.parse_obj TypeError: parse_obj() takes exactly 2 positional arguments (1 given) parse_raw():跟SwitchStrategy(**{"ids":[1,2,3]})类似，对参数要求为json参数
construct()：创建模型而不运行validator
validator
与drf-serializer类似，pydantic也提供对字段的单个校验和全部校验
- 单个字段的校验
class HttpUptimeCheckTaskConfig(BaseUptimeCheckTaskConfig):
@validator("method")
def validate_method(cls, value):
choices = [("GET", "GET"), ("POST", "POST"), ("PUT", "PUT"), ("PATCH", "PATCH"), ("DELETE", "DELETE")]
if value not in [k for k, v in choices]:
raise ValueError("invalid uptime check method")
return value
- validator复用
def validate_status(value):
if value not in [v for k, v in Status.__dict__.items() if not k.startswith("__")]:
raise ValueError("invalid uptime check task status")
return value
class ChangeUptimeCheckTaskStatus(BaseModel):
_validate_status: ClassVar[Callable] = validator("status", allow_reuse=True)(validate_status)
task_id: int = Field(description="任务ID")
status: str = Field(description="任务状态")
class ChangeUptimeCheckTaskStatusData(BaseModel):
_validate_status: ClassVar[Callable] = validator("status", allow_reuse=True)(validate_status)
id: int = Field(description="任务ID")
status: str = Field(description="任务状态")
- each_item使用(相当于for循环)
class A(BaseModel):
names:List[str]
@validator('names', each_item=True)
def check_names_not_empty(cls, v):
assert v != '', 'empty strings are not allowed.'
return v
A(names=["qq",""])
Traceback (most recent call last):
File "/Users/amiter/PycharmProjects/kmc/.venv/lib/python3.6/site-packages/IPython/core/interactiveshell.py", line 3343, in run_code
exec(code_obj, self.user_global_ns, self.user_ns)
File "<ipython-input-27-02207c301f6c>", line 1, in <module>
A(names=["qq",""])
File "pydantic/main.py", line 341, in pydantic.main.BaseModel.__init__
pydantic.error_wrappers.ValidationError: 1 validation error for A
names -> 1
empty strings are not allowed. (type=assertion_error)
注意:each_item不在从父类继承到的列表属性中使用,如下
class _A(BaseModel):
names:List[str]
class A(_A):
@validator('names', each_item=True)
def check_names_not_empty(cls, v):
assert v != '', 'empty strings are not allowed.'
return v
A(names=["qq",""])
Out[31]: A(names=['qq', ''])
- 整个模型验证
class RemoteCollectingSlz(BaseModel):
@root_validator(pre=True)
def check_remote_collecting_slz(cls, attrs: Dict) -> Dict:
assert "bk_host_id" in attrs or ("ip" in attrs and "bk_cloud_id" in attrs), "主机id和ip/bk_cloud_id不能同时为空"
return attrs
ip: Optional[str] = Field(description="远程采集服务器IP")
bk_cloud_id: Optional[int] = Field(description="蓝鲸云区域ID")
bk_host_id: Optional[int] = Field(description="主机ID")
bk_supplier_id: Optional[int] = Field(description="开发商账号")
is_collecting_only: bool = Field(description="是否只采集数据")
Model Config
可以通过Config模型来修改pydantic的行为
class Plugin(BaseModel):
@validator("plugin_id")
def validate_plugin_id(cls, v):
return v
class Config:
use_enum_values = True
validate_all = True
extra = Extra.allow
smart_union = True
os_type: OsType = Field(default=OsType.linux.value)
plugin_id: str = Field(default="")
from tewanwan import *
plugin = Plugin(strategy_id=12)
plugin.dict()
Out[4]: {'os_type': 'linux', 'plugin_id': '','strategy_id':12}
-
validate_all : 是否验证字段默认值，默认False,如下item_id默认写了字符串，并不会报错；
class Item(BaseModel): item_id:int = "hello" class Config: copy_on_model_validation = "shallow" # validate_all = True class Strategy(BaseModel): item: Item item = Item() strategy = Strategy(item=item)另外要注意如果在模型validator中定义了不存在就赋值的逻辑(创建时间，默认用户等)，就必须validate_all设 置为True -
use_enum_values:填充模型时是否采用枚举的属性，默认False,如果使用了枚举类型，建议开启
class OsType(str, Enum): linux = "linux" windows = "windows" class Plugin(BaseModel): os_type: OsType = Field(default=OsType.linux.value) plugin_id: str = Field(default="") Plugin().dict() Out[57]: {'os_type': <OsType.linux: 'linux'>, 'plugin_id': ''} Plugin.Config.use_enum_values = True Plugin().dict() Out[59]: {'os_type': 'linux', 'plugin_id': ''} extra:初始化期间是否忽略(
ignore)、允许(allow)或禁止(forbid)额外属性,默认是ignore''-
smart_union:避免union中的数据类型强转，默认False
from tewanwan import * Plugin(task_id=12).dict() Out[2]: {'os_type': 'linux', 'plugin_id': '', 'task_id': 12} Plugin.Config.smart_union = False Plugin(task_id=12).dict() Out[4]: {'os_type': 'linux', 'plugin_id': '', 'task_id': '12'} -
copy_on_model_validation:控制模型实例复制的方式，1.9.1版本默认深拷贝
-
'none'- 不做任何处理 -
'shallow'- 默认是浅拷贝 -
'deep'- 深拷贝
from pydantic import BaseModel class Item(BaseModel): item_id:int = 0 class Config: copy_on_model_validation = "none" class Strategy(BaseModel): item: Item item = Item() strategy = Strategy(item=item) assert strategy.item.item_id == 0 assert strategy.item.item_id is item.item_id assert strategy.item is item # Passes Item.Config.copy_on_model_validation="shallow" item = Item() strategy = Strategy(item=item) assert strategy.item is item Traceback (most recent call last): File "/Users/amiter/PycharmProjects/kmc/.venv/lib/python3.6/site-packages/IPython/core/interactiveshell.py", line 3343, in run_code exec(code_obj, self.user_global_ns, self.user_ns) File "<ipython-input-9-1a73d0a9a45e>", line 3, in <module> assert strategy.item is item AssertionError # Failed -
-
arbitrary_types_allowed: 是否允许用户自定义类型
from pydantic import BaseModel class Item(): def __init__(self, item_id: str): self.item_id = item_id class Strategy(BaseModel): item: Item class Config: arbitrary_types_allowed = True strategy = Strategy(item=Item(item_id='a_test')) -
validate_assignment:属性赋值时是否执行校验，默认为False，除非确定实例属性类型正确，否则建议开启
from pydantic import BaseModel class Item(BaseModel): item_id:int = 0 class Strategy(BaseModel): item: Item item = Item(item_id=10) item.item_id = {"sss"} strategy= Strategy(item=item) strategy.dict() Out[38]: {'item': {'item_id': {'sss'}}} Item.Config.validate_assignment = True item = Item(item_id=11) item.item_id = {"sss"} Traceback (most recent call last): File "/Users/amiter/PycharmProjects/kmc/.venv/lib/python3.6/site-packages/IPython/core/interactiveshell.py", line 3343, in run_code exec(code_obj, self.user_global_ns, self.user_ns) File "<ipython-input-41-d648006e3959>", line 1, in <module> item.item_id = {"sss"} File "pydantic/main.py", line 380, in pydantic.main.BaseModel.__setattr__ pydantic.error_wrappers.ValidationError: 1 validation error for Item item_id value is not a valid integer (type=type_error.integer)
注意点
模型之间继承后，序列化时会获取最先符合条件的模型
from pydantic import BaseModel as bs
from typing import Union
class A(bs):
a: int
class B(A):
b: int
class C(bs):
c: Union[B, A]
C(c={"a":1,"b":2}).dict()
Out[15]: {'c': {'a': 1}}
