Coverage for /usr/local/lib/python3.12/site-packages/prefect/_internal/lazy.py: 58%

1"""Utilities for lazy-loading module-level objects.""" 

2 

3from typing import Callable, Generic, TypeVar 1 ctx1a

4 

5K = TypeVar("K") 1 ctx1a

6V = TypeVar("V") 1 ctx1a

7 

8 

9class LazyDict(Generic[K, V]): 1 ctx1a

10 """ 

11 A dictionary-like object that defers loading its contents until first access. 

12 

13 Useful for module-level registries that import heavy dependencies. The loader 

14 function is called once on first access, and the result is cached. 

15 

16 Example: 

17 >>> def load_plugins() -> dict[str, type]: 

18 ... from heavy_module import PluginA, PluginB 

19 ... return {"a": PluginA, "b": PluginB} 

20 >>> 

21 >>> plugins: LazyDict[str, type] = LazyDict(load_plugins) 

22 >>> # Nothing loaded yet 

23 >>> plugins.get("a") # Triggers load, returns PluginA 

24 >>> plugins.get("b") # Uses cached result, returns PluginB 

25 """ 

26 

27 def __init__(self, loader_func: Callable[[], dict[K, V]]) -> None: 1 ctx1a

28 self._loader = loader_func 1 ctx1a

29 self._cache: dict[K, V] | None = None 1 ctx1a

30 

31 def _ensure_loaded(self) -> dict[K, V]: 1 ctx1a

32 """Load the dictionary contents if not already loaded.""" 

33 if self._cache is None: 

34 self._cache = self._loader() 

35 return self._cache 

36 

37 def get(self, key: K, default: V | None = None) -> V | None: 1 ctx1a

38 """Get an item from the lazy dict, loading if necessary.""" 

39 return self._ensure_loaded().get(key, default) 

40 

41 def __contains__(self, key: K) -> bool: 1 ctx1a

42 """Check if key is in the lazy dict, loading if necessary.""" 

43 return key in self._ensure_loaded() 

44 

45 def __getitem__(self, key: K) -> V: 1 ctx1a

46 """Get an item from the lazy dict, loading if necessary.""" 

47 return self._ensure_loaded()[key]