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

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 # This line should never be reached, but added for type checker 

76 raise RuntimeError("Unreachable code") 

77 

78 @internalcode 

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

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

81 

82 Returns: 

83 Sorted list of unique template names from all loaders 

84 """ 

85 self._ensure_initialized() 

86 

87 found_templates = set() 

88 

89 for loader in self.loaders: 

90 try: 

91 templates = await loader.list_templates_async() 

92 found_templates.update(templates) 

93 except (TypeError, NotImplementedError): 

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

95 continue 

96 except Exception: 

97 # Log error if debugging is enabled, but continue 

98 continue 

99 

100 return sorted(found_templates) 

101 

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

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

104 

105 Args: 

106 loader: Loader to add 

107 """ 

108 self.loaders.append(loader) 

109 

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

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

112 

113 Args: 

114 index: Position to insert at 

115 loader: Loader to insert 

116 """ 

117 self.loaders.insert(index, loader) 

118 

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

120 """Remove a loader from the sequence. 

121 

122 Args: 

123 loader: Loader to remove 

124 

125 Raises: 

126 ValueError: If loader is not in the sequence 

127 """ 

128 self.loaders.remove(loader) 

129 

130 def clear_loaders(self) -> None: 

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

132 self.loaders.clear() 

133 

134 def get_loader_count(self) -> int: 

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

136 

137 Returns: 

138 Number of loaders 

139 """ 

140 return len(self.loaders) 

141 

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

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

144 

145 Returns: 

146 Copy of the loader list 

147 """ 

148 return self.loaders.copy()