Coverage for jinja2_async_environment/loaders/filesystem.py: 63%

1"""Async filesystem template loader implementation.""" 

2 

3import typing as t 

4 

5from anyio import Path as AsyncPath 

6from jinja2.utils import internalcode 

7 

8from .base import AsyncBaseLoader, SourceType 

9 

10if t.TYPE_CHECKING: 

11 from ..environment import AsyncEnvironment 

12 

13 

14class AsyncFileSystemLoader(AsyncBaseLoader): 

15 """Async filesystem template loader with memory optimization. 

16 

17 This loader searches for templates in the filesystem asynchronously, 

18 supporting multiple search paths and proper file watching for 

19 template updates. 

20 """ 

21 

22 __slots__ = ("encoding", "followlinks", "_template_cache") 

23 

24 def __init__( 

25 self, 

26 searchpath: AsyncPath | str | t.Sequence[AsyncPath | str], 

27 encoding: str = "utf-8", 

28 followlinks: bool = False, 

29 ) -> None: 

30 """Initialize the filesystem loader. 

31 

32 Args: 

33 searchpath: Path or sequence of paths to search for templates 

34 encoding: File encoding to use when reading templates 

35 followlinks: Whether to follow symbolic links 

36 """ 

37 super().__init__(searchpath) 

38 self.encoding = encoding 

39 self.followlinks = followlinks 

40 self._template_cache: dict[str, tuple[float, str]] = {} 

41 

42 @internalcode 

43 async def get_source_async( 

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

45 ) -> SourceType: 

46 """Get template source from filesystem 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 template file cannot be found 

57 """ 

58 self._ensure_initialized() 

59 

60 for searchpath in self.searchpath: 

61 template_path = searchpath / name 

62 

63 try: 

64 if await template_path.exists(): 

65 if not await self._is_safe_path(template_path): 

66 continue 

67 

68 # Read the template file 

69 source = await template_path.read_text(encoding=self.encoding) 

70 filename = str(template_path.resolve()) 

71 

72 # Create uptodate function 

73 def uptodate() -> bool: 

74 try: 

75 return ( 

76 template_path.stat().st_mtime 

77 == template_path.stat().st_mtime 

78 ) 

79 except OSError: 

80 return False 

81 

82 return source, filename, uptodate 

83 

84 except OSError: 

85 # Continue to next search path 

86 continue 

87 

88 # Template not found in any search path 

89 self._handle_template_not_found(name) 

90 

91 async def _is_safe_path(self, template_path: AsyncPath) -> bool: 

92 """Check if the template path is safe to access. 

93 

94 Args: 

95 template_path: Path to check 

96 

97 Returns: 

98 True if path is safe, False otherwise 

99 """ 

100 try: 

101 # Check if it's a file 

102 if not await template_path.is_file(): 

103 return False 

104 

105 # If not following links, check for symlinks 

106 if not self.followlinks and await template_path.is_symlink(): 

107 return False 

108 

109 # Check that the resolved path is within search paths 

110 resolved_path = await template_path.resolve() 

111 

112 for searchpath in self.searchpath: 

113 resolved_searchpath = await searchpath.resolve() 

114 try: 

115 resolved_path.relative_to(resolved_searchpath) 

116 return True 

117 except ValueError: 

118 continue 

119 

120 return False 

121 

122 except OSError: 

123 return False 

124 

125 @internalcode 

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

127 """List all templates in the search paths asynchronously. 

128 

129 Returns: 

130 Sorted list of template names 

131 """ 

132 self._ensure_initialized() 

133 

134 found_templates = set() 

135 

136 for searchpath in self.searchpath: 

137 if not await searchpath.exists(): 

138 continue 

139 

140 try: 

141 # Use rglob to find all files recursively 

142 async for template_path in searchpath.rglob("*"): 

143 if await template_path.is_file(): 

144 # Get relative path from search path 

145 try: 

146 relative_path = template_path.relative_to(searchpath) 

147 template_name = str(relative_path).replace("\\", "/") 

148 

149 # Only include if it's a safe path 

150 if await self._is_safe_path(template_path): 

151 found_templates.add(template_name) 

152 

153 except ValueError: 

154 # Path is not relative to searchpath 

155 continue 

156 

157 except OSError: 

158 # Skip this search path if it can't be accessed 

159 continue 

160 

161 return sorted(found_templates) 

162 

163 async def _walk_directory(self, directory: AsyncPath): 

164 """Async generator to walk directory tree. 

165 

166 Args: 

167 directory: Directory to walk 

168 

169 Yields: 

170 AsyncPath objects for each file/directory found 

171 """ 

172 if not await directory.exists(): 

173 return 

174 

175 try: 

176 async for item in directory.iterdir(): 

177 yield item 

178 

179 if await item.is_dir() and ( 

180 self.followlinks or not await item.is_symlink() 

181 ): 

182 async for subitem in self._walk_directory(item): 

183 yield subitem 

184 

185 except (OSError, PermissionError): 

186 # Skip directories we can't access 

187 return 

188 

189 def _get_cache_key(self, name: str) -> str: 

190 """Generate cache key for template. 

191 

192 Args: 

193 name: Template name 

194 

195 Returns: 

196 Cache key string 

197 """ 

198 return f"fs:{name}"