因此,我正在为我的工作为 Azure 表创建一种 ORM,问题是,我不需要预先定义名为
table_client
的字段,因为这没有意义,我需要构建 table_client
在所有验证之后,因为它需要self.connection_string
和self.table_name
(下面的示例)。另外,我需要该属性在对象的所有存在之间保持持久性,因此每次调用方法或类似方法时我不会重新创建表客户端。所以我需要这个属性成为模型的一部分(即我需要 self.table_client
),问题是,我不能将其设置为可选或键入为 table_client: TableClient | None = None
,因为在下面,我收到了来自 linter 的一堆警告说“create_table”不是“None”的已知属性,或类似的属性。
这里的方法是什么?我不能使用
default_factory
,因为我显然需要经过验证的字段(connection_string
和 table_name
)(我认为)。这是我当前的代码(假设所有导入都是正确的):
class AzureTable(BaseModel):
connection_string: str
table_name: str
table_client: TableClient | None = None
def __post_init__(self):
self._create_connection()
def _create_connection(self):
self.table_client = TableServiceClient.from_connection_string( # Here is the freaking waning
conn_str=self.connection_string
).get_table_client(self.table_name)
return self
def create_table(self) -> None:
try:
self.table_client.create_table(table_name=self.table_name)
except HttpResponseError:
raise TableAlreadyExistsError()
def get_all(self) -> list:
return list(self.table_client.list_entities())
更多信息:
Python: 3.12
Pydantic: 2.8.2
azure-data-tables: 12.5.0
还尝试过:
class AzureTable(BaseModel):
connection_string: str
table_name: str
def __post_init__(self):
self._create_connection()
def _create_connection(self):
self.table_client = TableServiceClient.from_connection_string(
conn_str=self.connection_string
).get_table_client(self.table_name)
return self
def create_table(self) -> None:
try:
self.table_client.create_table(table_name=self.table_name)
except HttpResponseError:
raise TableAlreadyExistsError()
def get_all(self) -> list:
return list(self.table_client.list_entities())
但给了我一个错误:
File "/home/azureuser/dev/LangBotAPI/venv/lib/python3.12/site-packages/pydantic/main.py", line 828, in __getattr__ raise AttributeError(f'{type(self).__name__!r} object has no attribute {item!r}') AttributeError: 'BotTable' object has no attribute 'table_client'
也许您想要一个计算字段?看起来像这样:
from functools import cached_property
from pydantic import BaseModel, computed_field
class AzureTable(BaseModel):
connection_string: str
table_name: str
@computed_field
@cached_propery
def table_client(self) -> AppropriateReturnType:
table_client = (
TableServiceClient.from_connection_string(
conn_str=self.connection_string
).get_table_client(self.table_name)
)
return table_client
我对 Azure 模块一点也不熟悉(这就是为什么上面的示例用
AppropriateReturnType
注释),但这里有一个仅使用本机类型的可运行示例来展示它是如何工作的。鉴于以下情况:
from functools import cached_property
from pydantic import BaseModel, computed_field
class TableClient(BaseModel):
pass
class AzureTable(BaseModel):
connection_string: str
table_name: str
@computed_field
@cached_property
def table_client(self) -> str:
return f"{self.table_name}@{self.connection_string}"
我们可以这样做:
>>> x = AzureTable(connection_string='user@host', table_name='example')
>>> x
AzureTable(connection_string='user@host', table_name='example', table_client='example@user@host')
>>> x.table_client
'example@user@host'
请注意,因为我们使用的是
cached_property
装饰器,所以 table_client
仅计算一次。