Implementierung eines Lazy-evaluierten Felds für Pydantic v2
Posted: 16 Jan 2025, 11:35
Ich versuche, einen Lazily-evaluierten generischen Feldtyp für Pydantic v2 zu implementieren. Dies ist die einfache Implementierung, die ich habe. Sie können dem Lazy-Feld entweder einen Wert, eine Funktion oder eine asynchrone Funktion zuweisen und es wird nur ausgewertet, wenn Sie darauf zugreifen. Wenn Sie dies in einer normalen Klasse verwenden, funktioniert es perfekt. Aber es funktioniert nicht als Pydantic-Feld.
Das Problem ist, dass __set__ hier nie aufgerufen wird. __get__ wird jedoch aus irgendeinem Grund zweimal aufgerufen. Ich weiß, dass Pydantic intern einige seltsame Dinge tut, was der Grund dafür sein könnte. Für jede Hilfe zur Behebung dieses Problems wären wir sehr dankbar.
Dies ist die Ausgabe von oben:
Wie Sie sehen können, wird __get__ zweimal aufgerufen. Und weil __set__ nie aufgerufen wird, _is_loaded und _loader None ist, gibt __get__ einfach den Rohwert als Funktion ohne Auswertung zurück.
Das Problem ist, dass __set__ hier nie aufgerufen wird. __get__ wird jedoch aus irgendeinem Grund zweimal aufgerufen. Ich weiß, dass Pydantic intern einige seltsame Dinge tut, was der Grund dafür sein könnte. Für jede Hilfe zur Behebung dieses Problems wären wir sehr dankbar.
Code: Select all
import asyncio
import inspect
from typing import Any, Awaitable, Callable, Generic, Optional, TypeVar, Union, cast
from pydantic import BaseModel, GetCoreSchemaHandler
from pydantic_core import CoreSchema, core_schema
T = TypeVar("T")
class LazyField(Generic[T]):
"""A lazy field that can hold a value, function, or async function.
The value is evaluated only when accessed and then cached.
"""
def __init__(self, value=None) -> None:
print("LazyField.__init__")
self._value: Optional[T] = None
self._loader: Optional[Callable[[], Union[T, Awaitable[T]]]] = None
self._is_loaded: bool = False
def __get__(self, obj: Any, objtype=None) -> T:
print("LazyField.__get__")
if obj is None:
return self # type: ignore
if not self._is_loaded:
if self._loader is None:
if self._value is None:
raise AttributeError("LazyField has no value or loader set")
return self._value
if inspect.iscoroutinefunction(self._loader):
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
self._value = loop.run_until_complete(self._loader()) # type: ignore
else:
self._value = self._loader() # type: ignore
self._is_loaded = True
self._loader = None
assert self._value is not None
return self._value
def __set__(
self, obj: Any, value: Union[T, Callable[[], T], Callable[[], Awaitable[T]]]
) -> None:
print("LazyField.__set__")
self._is_loaded = False
if callable(value):
self._loader = cast(
Union[Callable[[], T], Callable[[], Awaitable[T]]], value
)
self._value = None
else:
self._loader = None
self._value = cast(T, value)
@classmethod
def __get_pydantic_core_schema__(
cls, source_type: type[Any], handler: GetCoreSchemaHandler
) -> CoreSchema:
print("LazyField.__get_pydantic_core_schema__")
# Extract the inner type from LazyField[T]
inner_type = (
source_type.__args__[0] if hasattr(source_type, "__args__") else Any
)
# Generate schema for the inner type
inner_schema = handler.generate_schema(inner_type)
schema = core_schema.json_or_python_schema(
json_schema=inner_schema,
python_schema=core_schema.union_schema(
[
# Handle direct value assignment
inner_schema,
# Handle callable assignment
core_schema.callable_schema(),
# Handle coroutine function assignment
core_schema.callable_schema(),
]
),
serialization=core_schema.plain_serializer_function_ser_schema(
lambda x: x._value if hasattr(x, "_value") and x._is_loaded else None,
return_schema=inner_schema,
when_used="json",
),
)
return schema
class A(BaseModel):
content: LazyField[bytes] = LazyField()
async def get_content():
return b"Hello, world!"
a = A(content=get_content)
print(a.content)
Code: Select all
LazyField.__init__
LazyField.__get__
LazyField.__get__
LazyField.__get_pydantic_core_schema__