Coverage for jinja2_async_environment/loaders/base.py: 76%
97 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-03 14:09 -0700
1"""Base classes and protocols for async template loaders."""
3import typing as t
4from abc import abstractmethod
6from anyio import Path as AsyncPath
7from jinja2.loaders import BaseLoader
8from jinja2.utils import internalcode
10if t.TYPE_CHECKING:
11 from jinja2 import Template
13 from ..environment import AsyncEnvironment
15# Type alias for source data
16SourceType = tuple[str, str | None, t.Callable[[], bool] | None]
19class AsyncLoaderProtocol(t.Protocol):
20 """Protocol defining the interface for async template loaders.
22 This protocol ensures all async loaders implement the required methods
23 for loading templates asynchronously while maintaining compatibility
24 with Jinja2's loader interface.
25 """
27 async def get_source_async(
28 self, environment: "AsyncEnvironment", name: str
29 ) -> SourceType | None:
30 """Get template source asynchronously.
32 Args:
33 environment: The async environment instance
34 name: Template name to load
36 Returns:
37 Tuple of (source, filename, uptodate_func) or None if not found
39 Raises:
40 TemplateNotFound: If template cannot be found
41 """
42 ...
44 async def list_templates_async(self) -> list[str]:
45 """List all available templates asynchronously.
47 Returns:
48 List of template names
50 Raises:
51 TypeError: If listing is not supported by this loader
52 """
53 ...
55 async def load_async(
56 self,
57 environment: "AsyncEnvironment",
58 name: str,
59 env_globals: dict[str, t.Any] | None = None,
60 ) -> "Template":
61 """Load template asynchronously.
63 Args:
64 environment: The async environment instance
65 name: Template name to load
66 env_globals: Global variables for the template
68 Returns:
69 Compiled Template object
71 Raises:
72 TemplateNotFound: If template cannot be found
73 """
74 ...
77class AsyncBaseLoader(BaseLoader):
78 """Base class for async template loaders with memory optimization.
80 This class provides the foundation for all async loaders, implementing
81 common functionality while using __slots__ for memory efficiency.
83 Features:
84 - Thread-safe lazy initialization
85 - Memory-optimized with __slots__
86 - Robust error handling and resource management
87 - Compatible with Jinja2's loader interface
88 - Support for async template operations
90 Thread Safety:
91 The loader implements thread-safe initialization using RLock to ensure
92 that multiple concurrent calls to _ensure_initialized() are handled safely.
93 """
95 __slots__ = ("searchpath", "_initialized", "_init_lock")
97 def __init__(self, searchpath: AsyncPath | str | t.Sequence[AsyncPath | str]):
98 """Initialize the async loader.
100 Args:
101 searchpath: Path or sequence of paths to search for templates
103 Raises:
104 TypeError: If searchpath is not a valid type
105 ValueError: If searchpath is empty or contains invalid paths
106 """
107 # Validate and normalize searchpath
108 searchpath_list: list[t.Any] = []
109 if isinstance(searchpath, str) or hasattr(
110 searchpath, "parts"
111 ): # AsyncPath check
112 searchpath_list = [searchpath]
113 else:
114 # Try to treat as sequence
115 try:
116 searchpath_list = list(searchpath) # type: ignore
117 if not searchpath_list:
118 raise ValueError("searchpath cannot be empty")
120 # Validate each path in the sequence
121 for i, path in enumerate(searchpath_list):
122 if not (isinstance(path, str) or hasattr(path, "parts")):
123 raise TypeError(
124 f"searchpath item {i} must be a string or AsyncPath, got {type(path)}"
125 )
126 except (TypeError, ValueError) as e:
127 if isinstance(e, ValueError):
128 raise
129 raise TypeError(
130 "searchpath must be a string, AsyncPath, or sequence of strings/AsyncPaths"
131 ) from e
133 # Convert to AsyncPath objects for consistency and validate
134 self.searchpath: list[AsyncPath] = []
135 for path in searchpath_list:
136 if isinstance(path, str):
137 if not path.strip():
138 raise ValueError("Empty string paths are not allowed")
139 self.searchpath.append(AsyncPath(path))
140 elif hasattr(path, "parts"):
141 # Already an AsyncPath-like object, convert to ensure type safety
142 self.searchpath.append(AsyncPath(str(path)))
143 else:
144 # Fallback: convert to string then AsyncPath
145 self.searchpath.append(AsyncPath(str(path)))
147 self._initialized = False
148 self._init_lock = None # Will be created on first use
150 def _ensure_initialized(self) -> None:
151 """Ensure the loader is properly initialized.
153 This method provides thread-safe lazy initialization. Subclasses
154 should override _perform_initialization() instead of this method.
156 Thread Safety:
157 Uses a reentrant lock (RLock) to ensure thread-safe initialization
158 even when called from multiple threads concurrently.
160 Raises:
161 RuntimeError: If initialization fails
162 """
163 if not self._initialized:
164 # Thread-safe initialization check
165 import threading
167 if not hasattr(self, "_init_lock") or self._init_lock is None:
168 self._init_lock = threading.RLock()
170 with self._init_lock:
171 if not self._initialized:
172 try:
173 self._perform_initialization()
174 self._initialized = True
175 except Exception:
176 # Ensure we don't leave loader in partially initialized state
177 self._initialized = False
178 # Re-raise the original exception to maintain test compatibility
179 raise
181 def _perform_initialization(self) -> None:
182 """Perform the actual initialization work.
184 This method should be overridden by subclasses that need
185 custom initialization logic. It is called within a thread-safe
186 context by _ensure_initialized().
188 Subclasses should implement their initialization logic here
189 rather than overriding _ensure_initialized() directly.
191 Raises:
192 Exception: Any initialization-specific exceptions
193 """
194 pass
196 @abstractmethod
197 async def get_source_async(
198 self, environment: "AsyncEnvironment", name: str
199 ) -> SourceType:
200 """Get template source asynchronously.
202 Must be implemented by subclasses.
204 Args:
205 environment: The async environment instance
206 name: Template name to load
208 Returns:
209 Tuple of (source, filename, uptodate_func)
211 Raises:
212 TemplateNotFound: If template cannot be found
213 """
214 raise NotImplementedError("Subclasses must implement get_source_async")
216 async def list_templates_async(self) -> list[str]:
217 """List all available templates asynchronously.
219 Default implementation raises TypeError. Override in subclasses
220 that support template listing.
222 Returns:
223 List of template names
225 Raises:
226 TypeError: If listing is not supported by this loader
227 """
228 raise TypeError("this loader cannot iterate over all templates")
230 @internalcode
231 async def load_async(
232 self,
233 environment: "AsyncEnvironment",
234 name: str,
235 env_globals: dict[str, t.Any] | None = None,
236 ) -> "Template":
237 """Load template asynchronously using get_source_async.
239 This method orchestrates the complete template loading process:
240 1. Get template source via get_source_async()
241 2. Handle bytecode caching if available
242 3. Compile template if not cached
243 4. Create and return Template instance
245 Args:
246 environment: The async environment instance
247 name: Template name to load
248 env_globals: Global variables for the template (optional)
250 Returns:
251 Compiled Template object ready for rendering
253 Raises:
254 TemplateNotFound: If template cannot be found
255 RuntimeError: If compilation or loading fails
256 """
257 if env_globals is None:
258 env_globals = {}
260 # Import TemplateNotFound here to avoid circular imports
261 from jinja2.exceptions import TemplateNotFound
263 # Validate inputs
264 if not name or not name.strip():
265 raise TemplateNotFound("Template name cannot be empty")
267 # Get template source
268 try:
269 source, path, uptodate = await self.get_source_async(environment, name)
270 except Exception as e:
271 if isinstance(e, TemplateNotFound):
272 raise
273 from jinja2.exceptions import TemplateNotFound as TNF
275 raise TNF(f"Failed to get template source: {e}") from e
277 # Normalize source to string
278 try:
279 source_str = source.decode("utf-8") if isinstance(source, bytes) else source
280 except UnicodeDecodeError as e:
281 from jinja2.exceptions import TemplateNotFound as TNF
283 raise TNF(f"Template {name} contains invalid UTF-8 encoding: {e}") from e
285 # Handle bytecode cache if available
286 bcc = environment.bytecode_cache
287 if bcc is not None:
288 try:
289 bucket = bcc.get_bucket(environment, name, path) # type: ignore
291 # Create checksum for bytecode caching
292 import hashlib
294 checksum = hashlib.sha256(source_str.encode("utf-8")).hexdigest()
296 # Try to get existing bytecode
297 code = bucket.get_bytecode(checksum)
299 if code is None:
300 # Compile template
301 code = environment.compile(source_str, name, path)
302 bucket.set_bytecode(checksum, code)
303 except Exception:
304 # Fallback to direct compilation if cache fails
305 code = environment.compile(source_str, name, path)
306 else:
307 # No cache, compile directly
308 code = environment.compile(source_str, name, path)
310 # Create template instance
311 template = environment.template_class.from_code(
312 environment,
313 code,
314 env_globals,
315 uptodate,
316 )
318 return template
320 def _get_cache_manager(self, environment: "AsyncEnvironment") -> t.Any:
321 """Get the cache manager from the environment.
323 Args:
324 environment: The async environment instance
326 Returns:
327 The cache manager instance
328 """
329 return getattr(environment, "cache_manager", None)
331 @internalcode
332 def _handle_template_not_found(self, name: str) -> None:
333 """Helper method to raise TemplateNotFound with consistent messaging.
335 Args:
336 name: Template name that was not found
338 Raises:
339 TemplateNotFound: Always raised with appropriate message
340 """
341 from jinja2.exceptions import TemplateNotFound
343 raise TemplateNotFound(name)