Coverage for src/lexigram/graphql/dataloader/registry.py: 100%

1"""DataLoaderProtocol Registry for per-request scoping. 

2 

3This module provides a registry for DataLoaderProtocol factories that creates 

4fresh loaders per request to prevent cache leakage. 

5""" 

6 

7from __future__ import annotations 

8 

9from collections.abc import Callable 

10 

11from lexigram.graphql.dataloader.loader import DataLoaderProtocol 

12from lexigram.primitives.registry import Registry 

13 

14 

15class DataLoaderRegistry(Registry[str, Callable[[], DataLoaderProtocol]]): 

16 """Registry for DataLoaderProtocol factories. 

17 

18 Creates fresh DataLoaders per request to prevent cache leakage 

19 between concurrent requests. Extends :class:`Registry` for 

20 unified introspection and lifecycle hooks. 

21 

22 Example: 

23 registry = DataLoaderRegistry() 

24 

25 def create_user_loader(context): 

26 return DataLoaderProtocol( 

27 name="users", 

28 batch_load_fn=lambda keys: fetch_users(keys), 

29 ) 

30 

31 registry.register("users", create_user_loader) 

32 

33 # In request context: 

34 loaders = registry.create_loaders(context) 

35 user_loader = loaders["users"] 

36 """ 

37 

38 def __init__(self) -> None: 

39 """Initialize the registry.""" 

40 super().__init__(name="graphql.dataloaders", allow_overwrite=True) 

41 

42 def create_loaders(self) -> dict[str, DataLoaderProtocol]: 

43 """Create all registered DataLoaders for a new request. 

44 

45 Returns: 

46 Dictionary mapping loader names to DataLoaderProtocol instances. 

47 """ 

48 return {name: factory() for name, factory in self.items()} 

49 

50 def get_names(self) -> list[str]: 

51 """Get all registered DataLoaderProtocol names. 

52 

53 Returns: 

54 List of DataLoaderProtocol names. 

55 """ 

56 return list(self.keys()) 

57 

58 

59# Decorator for easy registration 

60def dataloader( 

61 name: str, 

62) -> Callable[[Callable[[], DataLoaderProtocol]], Callable[[], DataLoaderProtocol]]: 

63 """Decorator to register a DataLoaderProtocol factory. 

64 

65 Args: 

66 name: Unique name for the DataLoaderProtocol. 

67 

68 Returns: 

69 Decorator function. 

70 

71 Example: 

72 @dataloader("users") 

73 def create_user_loader(): 

74 return DataLoaderProtocol( 

75 name="users", 

76 batch_load_fn=lambda keys: fetch_users(keys), 

77 ) 

78 """ 

79 

80 def decorator( 

81 factory: Callable[[], DataLoaderProtocol], 

82 ) -> Callable[[], DataLoaderProtocol]: 

83 # This will be registered via DataLoaderRegistry 

84 factory._dataloader_name = name # type: ignore[attr-defined] 

85 return factory 

86 

87 return decorator 

88 

89 

90__all__ = [ 

91 "DataLoaderRegistry", 

92 "dataloader", 

93]