Coverage for jinja2_async_environment/loaders/choice.py: 77%

1"""Async choice template loader implementation.""" 

2 

3import typing as t 

4 

5from anyio import Path as AsyncPath 

6from jinja2.exceptions import TemplateNotFound 

7from jinja2.utils import internalcode 

8 

9from .base import AsyncBaseLoader, AsyncLoaderProtocol, SourceType 

10 

11if t.TYPE_CHECKING: 

12 from ..environment import AsyncEnvironment 

13 

14 

15class AsyncChoiceLoader(AsyncBaseLoader): 

16 """Async choice template loader with memory optimization. 

17 

18 This loader tries multiple loaders in sequence until one successfully 

19 loads the requested template. Useful for fallback scenarios and 

20 template inheritance chains. 

21 """ 

22 

23 __slots__ = ("loaders",) 

24 

25 def __init__( 

26 self, 

27 loaders: t.Sequence[AsyncLoaderProtocol], 

28 searchpath: AsyncPath | str | t.Sequence[AsyncPath | str] | None = None, 

29 ) -> None: 

30 """Initialize the choice loader. 

31 

32 Args: 

33 loaders: Sequence of loaders to try in order 

34 searchpath: Path or sequence of paths for compatibility (not used) 

35 """ 

36 # Call parent with provided searchpath or empty list 

37 if searchpath is None: 

38 searchpath = [] 

39 super().__init__(searchpath) 

40 self.loaders = list(loaders) # Create a copy for safety 

41 

42 @internalcode 

43 async def get_source_async( 

44 self, environment: "AsyncEnvironment", name: str 

45 ) -> SourceType: 

46 """Get template source by trying loaders in sequence asynchronously. 

47 

48 Args: 

49 environment: The async environment instance 

50 name: Template name to load 

51 

52 Returns: 

53 Tuple of (source, filename, uptodate_func) 

54 

55 Raises: 

56 TemplateNotFound: If no loader can find the template 

57 """ 

58 self._ensure_initialized() 

59 

60 for loader in self.loaders: 

61 try: 

62 result = await loader.get_source_async(environment, name) 

63 if result is not None: 

64 return result 

65 except TemplateNotFound: 

66 # Try the next loader 

67 continue 

68 except Exception: 

69 # For other exceptions, continue to next loader 

70 # but log the error if debugging is enabled 

71 continue 

72 

73 # No loader could find the template 

74 self._handle_template_not_found(name) 

75 

76 @internalcode 

77 async def list_templates_async(self) -> list[str]: 

78 """List all templates from all loaders asynchronously. 

79 

80 Returns: 

81 Sorted list of unique template names from all loaders 

82 """ 

83 self._ensure_initialized() 

84 

85 found_templates = set() 

86 

87 for loader in self.loaders: 

88 try: 

89 templates = await loader.list_templates_async() 

90 found_templates.update(templates) 

91 except (TypeError, NotImplementedError): 

92 # Some loaders don't support listing, skip them 

93 continue 

94 except Exception: 

95 # Log error if debugging is enabled, but continue 

96 continue 

97 

98 return sorted(found_templates) 

99 

100 def add_loader(self, loader: AsyncLoaderProtocol) -> None: 

101 """Add a loader to the end of the sequence. 

102 

103 Args: 

104 loader: Loader to add 

105 """ 

106 self.loaders.append(loader) 

107 

108 def insert_loader(self, index: int, loader: AsyncLoaderProtocol) -> None: 

109 """Insert a loader at the specified position. 

110 

111 Args: 

112 index: Position to insert at 

113 loader: Loader to insert 

114 """ 

115 self.loaders.insert(index, loader) 

116 

117 def remove_loader(self, loader: AsyncLoaderProtocol) -> None: 

118 """Remove a loader from the sequence. 

119 

120 Args: 

121 loader: Loader to remove 

122 

123 Raises: 

124 ValueError: If loader is not in the sequence 

125 """ 

126 self.loaders.remove(loader) 

127 

128 def clear_loaders(self) -> None: 

129 """Remove all loaders from the sequence.""" 

130 self.loaders.clear() 

131 

132 def get_loader_count(self) -> int: 

133 """Get the number of loaders in the sequence. 

134 

135 Returns: 

136 Number of loaders 

137 """ 

138 return len(self.loaders) 

139 

140 def get_loaders(self) -> list[AsyncLoaderProtocol]: 

141 """Get a copy of the loader sequence. 

142 

143 Returns: 

144 Copy of the loader list 

145 """ 

146 return list(self.loaders)