Implementierung eines Lazy-evaluierten Felds für Pydantic v2 ⇐ Python

[Guest]

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.

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)

Dies ist die Ausgabe von oben:

Code: Select all

LazyField.__init__
LazyField.__get__
LazyField.__get__
LazyField.__get_pydantic_core_schema__

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.