pydantic 设置： 类型错误： 无法腌制'_thread.lock'对象

Question

我正面临pydantic.BaseSettings和prometheus_client.Summary的问题。下面的代码片段在试图执行时抛出异常:

from prometheus_client import Summary
from pydantic import BaseSettings
class Settings(BaseSettings):
app_name: str = 'any_app_name'
any_boolean: bool = False
any_summary: Summary = Summary('my_summary','My Summary')
s = Settings()

下面是stacktrace:

File "/Users/XXX/workspace/src/settings_summary.py", line 4, in <module>
class Settings(BaseSettings):
File "pydantic/main.py", line 252, in pydantic.main.ModelMetaclass.__new__
File "pydantic/fields.py", line 309, in pydantic.fields.ModelField.infer
File "pydantic/fields.py", line 271, in pydantic.fields.ModelField.__init__
File "pydantic/fields.py", line 344, in pydantic.fields.ModelField.prepare
File "pydantic/fields.py", line 368, in pydantic.fields.ModelField._set_default_and_type
File "pydantic/fields.py", line 280, in pydantic.fields.ModelField.get_default
File "/usr/local/Cellar/python@3.8/3.8.5/Frameworks/Python.framework/Versions/3.8/lib/python3.8/copy.py", line 172, in deepcopy
y = _reconstruct(x, memo, *rv)
File "/usr/local/Cellar/python@3.8/3.8.5/Frameworks/Python.framework/Versions/3.8/lib/python3.8/copy.py", line 270, in _reconstruct
state = deepcopy(state, memo)
File "/usr/local/Cellar/python@3.8/3.8.5/Frameworks/Python.framework/Versions/3.8/lib/python3.8/copy.py", line 146, in deepcopy
y = copier(x, memo)
File "/usr/local/Cellar/python@3.8/3.8.5/Frameworks/Python.framework/Versions/3.8/lib/python3.8/copy.py", line 230, in _deepcopy_dict
y[deepcopy(key, memo)] = deepcopy(value, memo)
File "/usr/local/Cellar/python@3.8/3.8.5/Frameworks/Python.framework/Versions/3.8/lib/python3.8/copy.py", line 172, in deepcopy
y = _reconstruct(x, memo, *rv)
File "/usr/local/Cellar/python@3.8/3.8.5/Frameworks/Python.framework/Versions/3.8/lib/python3.8/copy.py", line 270, in _reconstruct
state = deepcopy(state, memo)
File "/usr/local/Cellar/python@3.8/3.8.5/Frameworks/Python.framework/Versions/3.8/lib/python3.8/copy.py", line 146, in deepcopy
y = copier(x, memo)
File "/usr/local/Cellar/python@3.8/3.8.5/Frameworks/Python.framework/Versions/3.8/lib/python3.8/copy.py", line 230, in _deepcopy_dict
y[deepcopy(key, memo)] = deepcopy(value, memo)
File "/usr/local/Cellar/python@3.8/3.8.5/Frameworks/Python.framework/Versions/3.8/lib/python3.8/copy.py", line 161, in deepcopy
rv = reductor(4)
TypeError: cannot pickle '_thread.lock' object

删除Summary行代码时运行:

from pydantic import BaseSettings
class Settings(BaseSettings):
app_name: str = 'any_app_name'
any_boolean: bool = False
s = Settings()

你知道怎么修理它吗?

谢谢你提前…

Answer 1

看起来你想创建自定义数据类型

Answer 2

对我来说最简单的是使用一个自动排除的属性:

from typing import ClassVar
class Settings(BaseSettings):
any_summary: ClassVar[Summary] = Summary('my_summary','My Summary')

Answer 3

ClassVar不是最佳实践，因为它将构造一个将在所有实例之间共享的静态变量。

这是我推荐的解决方案，只是一个简单的演示，你可以类似地更新你的代码:

import threading
from typing import Optional, Any
from pydantic import BaseModel, Field
class User(BaseModel):
name: str = ...
family: str = ...
lock: Optional[type(threading.Lock())] = Field(  # use Optional to annotate the type and use Field to initialize
exclude=True,  # remember to set the exclude to True
title="lock"
)
rlock: Any = Field(  # or use Any to annotate the type and use Field to initialize
exclude=True,
title="rlock"
)
def __init__(self, **data):
super().__init__(**data)  # Pydantic init
self.lock = threading.Lock()  # remember to initialize inside the __init__ function
self.rlock = threading.RLock()
class Config:
arbitrary_types_allowed = True  # this flag must be set!

u = User(**{"name": "milad", "family": "vayani"})
print(u.json())