-A Python implementation of Liquid, the safe customer-facing template language for flexible web apps. +A Python engine for Liquid, the safe customer-facing template language for flexible web apps.
diff --git a/docs/loaders.md b/docs/loaders.md index 7f37b76d..50c92d53 100644 --- a/docs/loaders.md +++ b/docs/loaders.md @@ -1,5 +1,8 @@ # Template Loaders +::: liquid.loaders.CachingFileSystemLoader + handler: python + ::: liquid.loaders.FileSystemLoader handler: python diff --git a/liquid/__init__.py b/liquid/__init__.py index 244cc654..3bcdef4d 100644 --- a/liquid/__init__.py +++ b/liquid/__init__.py @@ -12,6 +12,7 @@ from .token import Token from .expression import Expression +from .loaders import CachingFileSystemLoader from .loaders import ChoiceLoader from .loaders import DictLoader from .loaders import FileExtensionLoader @@ -44,6 +45,7 @@ __all__ = ( "AwareBoundTemplate", "BoundTemplate", + "CachingFileSystemLoader", "ChoiceLoader", "Context", "ContextualTemplateAnalysis", diff --git a/liquid/builtin/loaders/__init__.py b/liquid/builtin/loaders/__init__.py new file mode 100644 index 00000000..3d08978f --- /dev/null +++ b/liquid/builtin/loaders/__init__.py @@ -0,0 +1,24 @@ +from .base_loader import BaseLoader +from .base_loader import DictLoader +from .base_loader import TemplateNamespace +from .base_loader import TemplateSource +from .base_loader import UpToDate + +from .choice_loader import ChoiceLoader + +from .file_system_loader import FileExtensionLoader +from .file_system_loader import FileSystemLoader + +from .caching_file_system_loader import CachingFileSystemLoader + +__all__ = ( + "BaseLoader", + "CachingFileSystemLoader", + "ChoiceLoader", + "DictLoader", + "FileExtensionLoader", + "FileSystemLoader", + "TemplateNamespace", + "TemplateSource", + "UpToDate", +) diff --git a/liquid/builtin/loaders/base_loader.py b/liquid/builtin/loaders/base_loader.py new file mode 100644 index 00000000..98bfdce4 --- /dev/null +++ b/liquid/builtin/loaders/base_loader.py @@ -0,0 +1,300 @@ +"""Base template loader.""" +from __future__ import annotations + +from abc import ABC +from pathlib import Path +from typing import TYPE_CHECKING +from typing import Awaitable +from typing import Callable +from typing import Dict +from typing import Mapping +from typing import NamedTuple +from typing import Optional +from typing import Union + +from liquid.exceptions import TemplateNotFound + +if TYPE_CHECKING: + from liquid import Context + from liquid import Environment + from liquid.template import BoundTemplate + +# ruff: noqa: D102 D101 + +TemplateNamespace = Optional[Mapping[str, object]] +UpToDate = Union[Callable[[], bool], Callable[[], Awaitable[bool]], None] + + +class TemplateSource(NamedTuple): + """A Liquid template source as returned by the `get_source` method of a `loader`. + + Attributes: + source: The liquid template source code. + filename: The liquid template file name or other string identifying its origin. + uptodate: Optional callable that will return `True` if the template is up to + date, or `False` if it needs to be reloaded. + matter: Optional mapping containing variables associated with the template. + Could be "front matter" or other meta data. + """ + + source: str + filename: str + uptodate: UpToDate + matter: TemplateNamespace = None + + +class BaseLoader(ABC): # noqa: B024 + """Base template loader from which all template loaders are derived. + + Attributes: + caching_loader (bool): Indicates if this loader implements its own cache. + Setting this sto `True` will cause the `Environment` to disable its cache + when initialized with a caching loader. + """ + + caching_loader = False + + def get_source( + self, + env: Environment, + template_name: str, + ) -> TemplateSource: + """Get the template source, filename and reload helper for a template. + + Args: + env: The `Environment` attempting to load the template source text. + template_name: A name or identifier for a template's source text. + """ + raise NotImplementedError("template loaders must implement a get_source method") + + async def get_source_async( + self, + env: Environment, + template_name: str, + ) -> TemplateSource: + """An async version of `get_source`. + + The default implementation delegates to `get_source()`. + """ + return self.get_source(env, template_name) + + def get_source_with_args( + self, + env: Environment, + template_name: str, + **kwargs: object, # noqa: ARG002 + ) -> TemplateSource: + """Get template source text, optionally referencing arbitrary keyword arguments. + + Keyword arguments can be useful for multi-user environments where you need to + modify a template loader's search space for a given user. + + By default, this method delegates to `get_source`, ignoring any keyword + arguments. + + _New in version 1.9.0._ + """ + return self.get_source(env, template_name) + + async def get_source_with_args_async( + self, env: Environment, template_name: str, **kwargs: object + ) -> TemplateSource: + """An async version of `get_source_with_args`. + + _New in version 1.9.0._ + """ + return self.get_source_with_args(env, template_name, **kwargs) + + def get_source_with_context( + self, + context: Context, + template_name: str, + **kwargs: str, # noqa: ARG002 + ) -> TemplateSource: + """Get a template's source, optionally referencing a render context.""" + return self.get_source(context.env, template_name) + + async def get_source_with_context_async( + self, + context: Context, + template_name: str, + **kwargs: str, # noqa: ARG002 + ) -> TemplateSource: + """An async version of `get_source_with_context`.""" + return await self.get_source_async(context.env, template_name) + + def load( + self, + env: Environment, + name: str, + globals: TemplateNamespace = None, # noqa: A002 + ) -> BoundTemplate: + """Load and parse a template. + + Used internally by `Environment` to load a template source. Delegates to + `get_source`. A custom loaders would typically implement `get_source` rather + than overriding `load`. + """ + try: + source, filename, uptodate, matter = self.get_source(env, name) + except Exception as err: # noqa: BLE001 + raise TemplateNotFound(name) from err + + template = env.from_string( + source, + globals=globals, + name=name, + path=Path(filename), + matter=matter, + ) + template.uptodate = uptodate + return template + + async def load_async( + self, + env: Environment, + name: str, + globals: TemplateNamespace = None, # noqa: A002 + ) -> BoundTemplate: + """An async version of `load`.""" + try: + template_source = await self.get_source_async(env, name) + source, filename, uptodate, matter = template_source + except Exception as err: # noqa: BLE001 + raise TemplateNotFound(name) from err + + template = env.from_string( + source, + globals=globals, + name=name, + path=Path(filename), + matter=matter, + ) + template.uptodate = uptodate + return template + + def load_with_args( + self, + env: Environment, + name: str, + globals: TemplateNamespace = None, # noqa: A002 + **kwargs: object, + ) -> BoundTemplate: + """Load template source text, optionally referencing extra keyword arguments. + + Most custom loaders will want to override `get_source_with_args()` rather than + this method. For example, you might want to override `load_with_args()` and + `get_source_with_args()` when implementing a custom caching loader. Where cache + handling happens in `load_*` methods. + """ + try: + source, filename, uptodate, matter = self.get_source_with_args( + env, name, **kwargs + ) + except Exception as err: # noqa: BLE001 + raise TemplateNotFound(name) from err + + template = env.from_string( + source, + globals=globals, + name=name, + path=Path(filename), + matter=matter, + ) + template.uptodate = uptodate + return template + + async def load_with_args_async( + self, + env: Environment, + name: str, + globals: TemplateNamespace = None, # noqa: A002 + **kwargs: object, + ) -> BoundTemplate: + """An async version of `load_with_args()`.""" + try: + template_source = await self.get_source_with_args_async(env, name, **kwargs) + source, filename, uptodate, matter = template_source + except Exception as err: # noqa: BLE001 + raise TemplateNotFound(name) from err + + template = env.from_string( + source, + globals=globals, + name=name, + path=Path(filename), + matter=matter, + ) + template.uptodate = uptodate + return template + + def load_with_context( + self, + context: Context, + name: str, + **kwargs: str, + ) -> BoundTemplate: + """Load and parse a template, optionally referencing a render context.""" + try: + source, filename, uptodate, matter = self.get_source_with_context( + context, name, **kwargs + ) + except Exception as err: # noqa: BLE001 + raise TemplateNotFound(name) from err + + template = context.env.from_string( + source, + globals=context.globals, + name=name, + path=Path(filename), + matter=matter, + ) + template.uptodate = uptodate + return template + + async def load_with_context_async( + self, + context: Context, + name: str, + **kwargs: str, + ) -> BoundTemplate: + """An async version of `load_with_context`.""" + try: + ( + source, + filename, + uptodate, + matter, + ) = await self.get_source_with_context_async(context, name, **kwargs) + except Exception as err: # noqa: BLE001 + raise TemplateNotFound(name) from err + + template = context.env.from_string( + source, + globals=context.globals, + name=name, + path=Path(filename), + matter=matter, + ) + template.uptodate = uptodate + return template + + +class DictLoader(BaseLoader): + """A loader that loads templates from a dictionary. + + Args: + templates: A dictionary mapping template names to template source strings. + """ + + def __init__(self, templates: Dict[str, str]): + super().__init__() + self.templates = templates + + def get_source(self, _: Environment, template_name: str) -> TemplateSource: + try: + source = self.templates[template_name] + except KeyError as err: + raise TemplateNotFound(template_name) from err + + return TemplateSource(source, template_name, None) diff --git a/liquid/builtin/loaders/caching_file_system_loader.py b/liquid/builtin/loaders/caching_file_system_loader.py new file mode 100644 index 00000000..9b945273 --- /dev/null +++ b/liquid/builtin/loaders/caching_file_system_loader.py @@ -0,0 +1,228 @@ +"""A file system loader that caches parsed templates in memory.""" +from __future__ import annotations + +from functools import partial +from typing import TYPE_CHECKING +from typing import Awaitable +from typing import Callable +from typing import Iterable +from typing import Mapping +from typing import Union + +from liquid.utils import LRUCache + +from .file_system_loader import FileExtensionLoader + +if TYPE_CHECKING: + from pathlib import Path + + from liquid import BoundTemplate + from liquid import Context + from liquid import Environment + + from .base_loader import TemplateNamespace + from .base_loader import TemplateSource + +# ruff: noqa: D102 D101 + + +class CachingFileSystemLoader(FileExtensionLoader): + """A file system loader that caches parsed templates in memory. + + Args: + search_path: One or more paths to search. + encoding: Open template files with the given encoding. + ext: A default file extension. Should include a leading period. + auto_reload: If `True`, automatically reload a cached template if it has been + updated. + namespace_key: The name of a global render context variable or loader keyword + argument that resolves to the current loader "namespace" or "scope". + + If you're developing a multi-user application, a good namespace might be + `uid`, where `uid` is a unique identifier for a user and templates are + arranged in folders named for each `uid` inside the search path. + cache_size: The maximum number of templates to hold in the cache before removing + the least recently used template. + """ + + caching_loader = True + + def __init__( + self, + search_path: Union[str, Path, Iterable[Union[str, Path]]], + encoding: str = "utf-8", + ext: str = ".liquid", + *, + auto_reload: bool = True, + namespace_key: str = "", + cache_size: int = 300, + ): + super().__init__( + search_path=search_path, + encoding=encoding, + ext=ext, + ) + self.auto_reload = auto_reload + self.cache = LRUCache(capacity=cache_size) + self.namespace_key = namespace_key + + def load( + self, + env: Environment, + name: str, + globals: TemplateNamespace = None, # noqa: A002 + ) -> BoundTemplate: + return self.check_cache( + env, + name, + globals, + partial(super().load, env, name, globals), + ) + + async def load_async( + self, + env: Environment, + name: str, + globals: TemplateNamespace = None, # noqa: A002 + ) -> BoundTemplate: + return await self.check_cache_async( + env, + name, + globals, + partial(super().load_async, env, name, globals), + ) + + def load_with_args( + self, + env: Environment, + name: str, + globals: TemplateNamespace = None, # noqa: A002 + **kwargs: object, + ) -> BoundTemplate: + cache_key = self.cache_key(name, kwargs) + return self.check_cache( + env, + cache_key, + globals, + partial(super().load_with_args, env, cache_key, globals, **kwargs), + ) + + async def load_with_args_async( + self, + env: Environment, + name: str, + globals: TemplateNamespace = None, # noqa: A002 + **kwargs: object, + ) -> BoundTemplate: + cache_key = self.cache_key(name, kwargs) + return await self.check_cache_async( + env, + cache_key, + globals, + partial(super().load_with_args_async, env, cache_key, globals, **kwargs), + ) + + def load_with_context( + self, context: Context, name: str, **kwargs: str + ) -> BoundTemplate: + cache_key = self.cache_key_with_context(name, context, **kwargs) + return self.check_cache( + context.env, + cache_key, + context.globals, + partial(super().load_with_context, context=context, name=name, **kwargs), + ) + + async def load_with_context_async( + self, context: Context, name: str, **kwargs: str + ) -> BoundTemplate: + cache_key = self.cache_key_with_context(name, context, **kwargs) + return await self.check_cache_async( + context.env, + cache_key, + context.globals, + partial(super().load_with_context_async, context, name, **kwargs), + ) + + def check_cache( + self, + env: Environment, # noqa: ARG002 + cache_key: str, + globals: TemplateNamespace, # noqa: A002 + load_func: Callable[[], BoundTemplate], + ) -> BoundTemplate: + try: + cached_template: BoundTemplate = self.cache[cache_key] + except KeyError: + template = load_func() + self.cache[cache_key] = template + return template + + if self.auto_reload and not cached_template.is_up_to_date: + template = load_func() + self.cache[cache_key] = template + return template + + if globals: + cached_template.globals.update(globals) + return cached_template + + async def check_cache_async( + self, + env: Environment, # noqa: ARG002 + cache_key: str, + globals: TemplateNamespace, # noqa: A002 + load_func: Callable[[], Awaitable[BoundTemplate]], + ) -> BoundTemplate: + try: + cached_template: BoundTemplate = self.cache[cache_key] + except KeyError: + template = await load_func() + self.cache[cache_key] = template + return template + + if self.auto_reload and not await cached_template.is_up_to_date_async(): + template = await load_func() + self.cache[cache_key] = template + return template + + if globals: + cached_template.globals.update(globals) + return cached_template + + def cache_key(self, name: str, args: Mapping[str, object]) -> str: + if not self.namespace_key: + return name + + try: + return f"{args[self.namespace_key]}/{name}" + except KeyError: + return name + + def cache_key_with_context( + self, + name: str, + context: Context, + **kwargs: str, # noqa: ARG002 + ) -> str: + if not self.namespace_key: + return name + + try: + return f"{context.globals[self.namespace_key]}/{name}" + except KeyError: + return name + + def get_source_with_context( + self, context: Context, template_name: str, **kwargs: str + ) -> TemplateSource: + # In this case, our cache key and real file name are the same. + name = self.cache_key_with_context(template_name, context, **kwargs) + return self.get_source(context.env, name) + + async def get_source_with_context_async( + self, context: Context, template_name: str, **kwargs: str + ) -> TemplateSource: + # In this case, our cache key and real file name are the same. + name = self.cache_key_with_context(template_name, context, **kwargs) + return await self.get_source_async(context.env, name) diff --git a/liquid/builtin/loaders/choice_loader.py b/liquid/builtin/loaders/choice_loader.py new file mode 100644 index 00000000..47495416 --- /dev/null +++ b/liquid/builtin/loaders/choice_loader.py @@ -0,0 +1,49 @@ +"""A template loader that delegates to other template loaders.""" +from __future__ import annotations + +from typing import TYPE_CHECKING +from typing import List + +from liquid.exceptions import TemplateNotFound + +from .base_loader import BaseLoader +from .base_loader import TemplateSource + +if TYPE_CHECKING: + from liquid import Environment + + +class ChoiceLoader(BaseLoader): + """A template loader that will try each of a list of loaders in turn. + + Args: + loaders: A list of loaders implementing `liquid.loaders.BaseLoader`. + """ + + def __init__(self, loaders: List[BaseLoader]): + super().__init__() + self.loaders = loaders + + def get_source( # noqa: D102 + self, env: Environment, template_name: str + ) -> TemplateSource: + for loader in self.loaders: + try: + return loader.get_source(env, template_name) + except TemplateNotFound: + pass + + raise TemplateNotFound(template_name) + + async def get_source_async( # noqa: D102 + self, + env: Environment, + template_name: str, + ) -> TemplateSource: + for loader in self.loaders: + try: + return await loader.get_source_async(env, template_name) + except TemplateNotFound: + pass + + raise TemplateNotFound(template_name) diff --git a/liquid/builtin/loaders/file_system_loader.py b/liquid/builtin/loaders/file_system_loader.py new file mode 100644 index 00000000..396cbfff --- /dev/null +++ b/liquid/builtin/loaders/file_system_loader.py @@ -0,0 +1,132 @@ +"""Built-in file system loader.""" +from __future__ import annotations + +import asyncio +import os +from functools import partial +from pathlib import Path +from typing import TYPE_CHECKING +from typing import Iterable +from typing import Tuple +from typing import Union + +from liquid.exceptions import TemplateNotFound + +from .base_loader import BaseLoader +from .base_loader import TemplateSource + +if TYPE_CHECKING: + from liquid import Environment + + +class FileSystemLoader(BaseLoader): + """A loader that loads templates from one or more directories on the file system. + + Args: + search_path: One or more paths to search. + encoding: Open template files with the given encoding. + """ + + def __init__( + self, + search_path: Union[str, Path, Iterable[Union[str, Path]]], + encoding: str = "utf-8", + ): + super().__init__() + if not isinstance(search_path, Iterable) or isinstance(search_path, str): + search_path = [search_path] + + self.search_path = [Path(path) for path in search_path] + self.encoding = encoding + + def resolve_path(self, template_name: str) -> Path: + """Return a path to the template `template_name`. + + If the search path is a list of paths, returns the first path where + `template_name` exists. If none of the search paths contain `template_name`, a + `TemplateNotFound` exception is raised. + """ + template_path = Path(template_name) + + if os.path.pardir in template_path.parts: + raise TemplateNotFound(template_name) + + for path in self.search_path: + source_path = path.joinpath(template_path) + if not source_path.exists(): + continue + return source_path + raise TemplateNotFound(template_name) + + def _read(self, source_path: Path) -> Tuple[str, float]: + with source_path.open(encoding=self.encoding) as fd: + source = fd.read() + return source, source_path.stat().st_mtime + + def get_source( # noqa: D102 + self, _: Environment, template_name: str + ) -> TemplateSource: + source_path = self.resolve_path(template_name) + source, mtime = self._read(source_path) + return TemplateSource( + source, + str(source_path), + partial(self._uptodate, source_path, mtime), + ) + + @staticmethod + def _uptodate(source_path: Path, mtime: float) -> bool: + return mtime == source_path.stat().st_mtime + + @staticmethod + async def _uptodate_async(source_path: Path, mtime: float) -> bool: + return await asyncio.get_running_loop().run_in_executor( + None, lambda: mtime == source_path.stat().st_mtime + ) + + async def get_source_async( # noqa: D102 + self, _: Environment, template_name: str + ) -> TemplateSource: + loop = asyncio.get_running_loop() + source_path = await loop.run_in_executor(None, self.resolve_path, template_name) + source, mtime = await loop.run_in_executor(None, self._read, source_path) + return TemplateSource( + source, str(source_path), partial(self._uptodate_async, source_path, mtime) + ) + + +class FileExtensionLoader(FileSystemLoader): + """A file system loader that adds a file name extension if one is missing. + + Args: + search_path: One or more paths to search. + encoding: Open template files with the given encoding. + ext: A default file extension. Should include a leading period. + """ + + def __init__( + self, + search_path: Union[str, Path, Iterable[Union[str, Path]]], + encoding: str = "utf-8", + ext: str = ".liquid", + ): + super().__init__(search_path=search_path, encoding=encoding) + self.ext = ext + + def resolve_path(self, template_name: str) -> Path: # noqa: D102 + template_path = Path(template_name) + + if not template_path.suffix: + template_path = template_path.with_suffix(self.ext) + + # Don't allow "../" to escape the search path. + if os.path.pardir in template_path.parts: + raise TemplateNotFound(template_name) + + for path in self.search_path: + source_path = path.joinpath(template_path) + + if not source_path.exists(): + continue + return source_path + raise TemplateNotFound(template_name) diff --git a/liquid/environment.py b/liquid/environment.py index ee5ece63..dff1b4be 100644 --- a/liquid/environment.py +++ b/liquid/environment.py @@ -1,5 +1,4 @@ """Shared configuration from which templates can be loaded and parsed.""" - from __future__ import annotations import warnings @@ -219,7 +218,7 @@ def __init__( self.mode = tolerance # Template cache - if cache_size and cache_size > 0: + if cache_size and cache_size > 0 and not self.loader.caching_loader: self.cache: Optional[MutableMapping[Any, Any]] = LRUCache(cache_size) self.auto_reload = auto_reload else: diff --git a/liquid/loaders.py b/liquid/loaders.py index 62e4e5cd..2cce16f1 100644 --- a/liquid/loaders.py +++ b/liquid/loaders.py @@ -1,442 +1,22 @@ -"""Base class and file system implementation for loading template sources.""" -from __future__ import annotations - -import asyncio -import os -from abc import ABC -from collections import abc -from functools import partial -from pathlib import Path -from typing import TYPE_CHECKING -from typing import Awaitable -from typing import Callable -from typing import Dict -from typing import Iterable -from typing import List -from typing import Mapping -from typing import NamedTuple -from typing import Optional -from typing import Tuple -from typing import Union - -from liquid.exceptions import TemplateNotFound - -if TYPE_CHECKING: - from liquid import Context - from liquid import Environment - from liquid.template import BoundTemplate - -# ruff: noqa: D102 D101 - -UpToDate = Union[Callable[[], bool], Callable[[], Awaitable[bool]], None] - - -class TemplateSource(NamedTuple): - """A Liquid template source as returned by the `get_source` method of a `loader`. - - Attributes: - source: The liquid template source code. - filename: The liquid template file name or other string identifying its origin. - uptodate: Optional callable that will return `True` if the template is up to - date, or `False` if it needs to be reloaded. - matter: Optional mapping containing variables associated with the template. - Could be "front matter" or other meta data. - """ - - source: str - filename: str - uptodate: UpToDate - matter: Optional[Mapping[str, object]] = None - - -class BaseLoader(ABC): # noqa: B024 - """Base template loader from which all template loaders are derived.""" - - def get_source( - self, - env: Environment, - template_name: str, - ) -> TemplateSource: - """Get the template source, filename and reload helper for a template. - - Args: - env: The `Environment` attempting to load the template source text. - template_name: A name or identifier for a template's source text. - """ - raise NotImplementedError("template loaders must implement a get_source method") - - async def get_source_async( - self, - env: Environment, - template_name: str, - ) -> TemplateSource: - """An async version of `get_source`. - - The default implementation delegates to `get_source()`. - """ - return self.get_source(env, template_name) - - def get_source_with_args( - self, - env: Environment, - template_name: str, - **kwargs: object, # noqa: ARG002 - ) -> TemplateSource: - """Get template source text, optionally referencing arbitrary keyword arguments. - - Keyword arguments can be useful for multi-user environments where you need to - modify a template loader's search space for a given user. - - By default, this method delegates to `get_source`, ignoring any keyword - arguments. - - _New in version 1.9.0._ - """ - return self.get_source(env, template_name) - - async def get_source_with_args_async( - self, env: Environment, template_name: str, **kwargs: object - ) -> TemplateSource: - """An async version of `get_source_with_args`. - - _New in version 1.9.0._ - """ - return self.get_source_with_args(env, template_name, **kwargs) - - def get_source_with_context( - self, - context: Context, - template_name: str, - **kwargs: str, # noqa: ARG002 - ) -> TemplateSource: - """Get a template's source, optionally referencing a render context.""" - return self.get_source(context.env, template_name) - - async def get_source_with_context_async( - self, - context: Context, - template_name: str, - **kwargs: str, # noqa: ARG002 - ) -> TemplateSource: - """An async version of `get_source_with_context`.""" - return await self.get_source_async(context.env, template_name) - - def load( - self, - env: Environment, - name: str, - globals: Optional[Mapping[str, object]] = None, # noqa: A002 - ) -> BoundTemplate: - """Load and parse a template. - - Used internally by `Environment` to load a template source. Delegates to - `get_source`. A custom loaders would typically implement `get_source` rather - than overriding `load`. - """ - try: - source, filename, uptodate, matter = self.get_source(env, name) - except Exception as err: # noqa: BLE001 - raise TemplateNotFound(name) from err - - template = env.from_string( - source, - globals=globals, - name=name, - path=Path(filename), - matter=matter, - ) - template.uptodate = uptodate - return template - - async def load_async( - self, - env: Environment, - name: str, - globals: Optional[Mapping[str, object]] = None, # noqa: A002 - ) -> BoundTemplate: - """An async version of `load`.""" - try: - template_source = await self.get_source_async(env, name) - source, filename, uptodate, matter = template_source - except Exception as err: # noqa: BLE001 - raise TemplateNotFound(name) from err - - template = env.from_string( - source, - globals=globals, - name=name, - path=Path(filename), - matter=matter, - ) - template.uptodate = uptodate - return template - - def load_with_args( - self, - env: Environment, - name: str, - globals: Optional[Mapping[str, object]] = None, # noqa: A002 - **kwargs: object, - ) -> BoundTemplate: - """Load template source text, optionally referencing extra keyword arguments. - - Most custom loaders will want to override `get_source_with_args()` rather than - this method. For example, you might want to override `load_with_args()` and - `get_source_with_args()` when implementing a custom caching loader. Where cache - handling happens in `load_*` methods. - """ - try: - source, filename, uptodate, matter = self.get_source_with_args( - env, name, **kwargs - ) - except Exception as err: # noqa: BLE001 - raise TemplateNotFound(name) from err - - template = env.from_string( - source, - globals=globals, - name=name, - path=Path(filename), - matter=matter, - ) - template.uptodate = uptodate - return template - - async def load_with_args_async( - self, - env: Environment, - name: str, - globals: Optional[Mapping[str, object]] = None, # noqa: A002 - **kwargs: object, - ) -> BoundTemplate: - """An async version of `load_with_args()`.""" - try: - template_source = await self.get_source_with_args_async(env, name, **kwargs) - source, filename, uptodate, matter = template_source - except Exception as err: # noqa: BLE001 - raise TemplateNotFound(name) from err - - template = env.from_string( - source, - globals=globals, - name=name, - path=Path(filename), - matter=matter, - ) - template.uptodate = uptodate - return template - - def load_with_context( - self, - context: Context, - name: str, - **kwargs: str, - ) -> BoundTemplate: - """Load and parse a template, optionally referencing a render context.""" - try: - source, filename, uptodate, matter = self.get_source_with_context( - context, name, **kwargs - ) - except Exception as err: # noqa: BLE001 - raise TemplateNotFound(name) from err - - template = context.env.from_string( - source, - globals=context.globals, - name=name, - path=Path(filename), - matter=matter, - ) - template.uptodate = uptodate - return template - - async def load_with_context_async( - self, - context: Context, - name: str, - **kwargs: str, - ) -> BoundTemplate: - """An async version of `load_with_context`.""" - try: - ( - source, - filename, - uptodate, - matter, - ) = await self.get_source_with_context_async(context, name, **kwargs) - except Exception as err: # noqa: BLE001 - raise TemplateNotFound(name) from err - - template = context.env.from_string( - source, - globals=context.globals, - name=name, - path=Path(filename), - matter=matter, - ) - template.uptodate = uptodate - return template - - -class FileSystemLoader(BaseLoader): - """A loader that loads templates from one or more directories on the file system. - - Args: - search_path: One or more paths to search. - encoding: Open template files with the given encoding. - """ - - def __init__( - self, - search_path: Union[str, Path, Iterable[Union[str, Path]]], - encoding: str = "utf-8", - ): - if not isinstance(search_path, abc.Iterable) or isinstance(search_path, str): - search_path = [search_path] - - self.search_path = [Path(path) for path in search_path] - self.encoding = encoding - - def resolve_path(self, template_name: str) -> Path: - """Return a path to the template `template_name`. - - If the search path is a list of paths, returns the first path where - `template_name` exists. If none of the search paths contain `template_name`, a - `TemplateNotFound` exception is raised. - """ - template_path = Path(template_name) - - if os.path.pardir in template_path.parts: - raise TemplateNotFound(template_name) - - for path in self.search_path: - source_path = path.joinpath(template_path) - if not source_path.exists(): - continue - return source_path - raise TemplateNotFound(template_name) - - def _read(self, source_path: Path) -> Tuple[str, float]: - with source_path.open(encoding=self.encoding) as fd: - source = fd.read() - return source, source_path.stat().st_mtime - - def get_source(self, _: Environment, template_name: str) -> TemplateSource: - source_path = self.resolve_path(template_name) - source, mtime = self._read(source_path) - return TemplateSource( - source, - str(source_path), - partial(self._uptodate, source_path, mtime), - ) - - @staticmethod - def _uptodate(source_path: Path, mtime: float) -> bool: - return mtime == source_path.stat().st_mtime - - @staticmethod - async def _uptodate_async(source_path: Path, mtime: float) -> bool: - return await asyncio.get_running_loop().run_in_executor( - None, lambda: mtime == source_path.stat().st_mtime - ) - - async def get_source_async( - self, _: Environment, template_name: str - ) -> TemplateSource: - loop = asyncio.get_running_loop() - source_path = await loop.run_in_executor(None, self.resolve_path, template_name) - source, mtime = await loop.run_in_executor(None, self._read, source_path) - return TemplateSource( - source, str(source_path), partial(self._uptodate_async, source_path, mtime) - ) - - -class FileExtensionLoader(FileSystemLoader): - """A file system loader that adds a file name extension if one is missing. - - : Args: - search_path: One or more paths to search. - encoding: Open template files with the given encoding. - ext: A default file extension. Should include a leading period. - """ - - def __init__( - self, - search_path: Union[str, Path, Iterable[Union[str, Path]]], - encoding: str = "utf-8", - ext: str = ".liquid", - ): - super().__init__(search_path, encoding=encoding) - self.ext = ext - - def resolve_path(self, template_name: str) -> Path: - template_path = Path(template_name) - - if not template_path.suffix: - template_path = template_path.with_suffix(self.ext) - - # Don't allow "../" to escape the search path. - if os.path.pardir in template_path.parts: - raise TemplateNotFound(template_name) - - for path in self.search_path: - source_path = path.joinpath(template_path) - - if not source_path.exists(): - continue - return source_path - raise TemplateNotFound(template_name) - - -class DictLoader(BaseLoader): - """A loader that loads templates from a dictionary. - - Args: - templates: A dictionary mapping template names to template source strings. - """ - - def __init__(self, templates: Dict[str, str]): - super().__init__() - self.templates = templates - - def get_source(self, _: Environment, template_name: str) -> TemplateSource: - try: - source = self.templates[template_name] - except KeyError as err: - raise TemplateNotFound(template_name) from err - - return TemplateSource(source, template_name, None) - - -class ChoiceLoader(BaseLoader): - """A template loader that will try each of a list of loaders in turn. - - Args: - loaders: A list of loaders implementing `liquid.loaders.BaseLoader`. - """ - - def __init__(self, loaders: List[BaseLoader]): - super().__init__() - self.loaders = loaders - - def get_source(self, env: Environment, template_name: str) -> TemplateSource: - for loader in self.loaders: - try: - return loader.get_source(env, template_name) - except TemplateNotFound: - pass - - raise TemplateNotFound(template_name) - - async def get_source_async( - self, - env: Environment, - template_name: str, - ) -> TemplateSource: - for loader in self.loaders: - try: - return await loader.get_source_async(env, template_name) - except TemplateNotFound: - pass - - raise TemplateNotFound(template_name) +"""Built-in loaders.""" +from .builtin.loaders import BaseLoader +from .builtin.loaders import CachingFileSystemLoader +from .builtin.loaders import ChoiceLoader +from .builtin.loaders import DictLoader +from .builtin.loaders import FileExtensionLoader +from .builtin.loaders import FileSystemLoader +from .builtin.loaders import TemplateNamespace +from .builtin.loaders import TemplateSource +from .builtin.loaders import UpToDate + +__all__ = ( + "BaseLoader", + "CachingFileSystemLoader", + "ChoiceLoader", + "DictLoader", + "FileExtensionLoader", + "FileSystemLoader", + "TemplateNamespace", + "TemplateSource", + "UpToDate", +) diff --git a/pyproject.toml b/pyproject.toml index e53924b5..eed190a4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -209,6 +209,7 @@ convention = "google" "liquid/__about__.py" = ["D100"] "liquid/__init__.py" = ["D104", "I001"] "liquid/builtin/filters/__init__.py" = ["D104", "I001"] +"liquid/builtin/loaders/__init__.py" = ["D104", "I001"] "liquid/builtin/tags/__init__.py" = ["D104", "I001"] "tests/*" = ["D100", "D101", "D104", "D103", "D102", "D209", "D205", "SIM117"] "scripts/__init__.py" = ["D104", "I001"] diff --git a/tests/fixtures/namespace/dropify/index.liquid b/tests/fixtures/namespace/dropify/index.liquid new file mode 100644 index 00000000..89bda4ac --- /dev/null +++ b/tests/fixtures/namespace/dropify/index.liquid @@ -0,0 +1 @@ +namespaced template \ No newline at end of file diff --git a/tests/test_caching_loader.py b/tests/test_caching_loader.py new file mode 100644 index 00000000..7622aa76 --- /dev/null +++ b/tests/test_caching_loader.py @@ -0,0 +1,238 @@ +import asyncio +import tempfile +import time +import unittest +from pathlib import Path + +from liquid import BoundTemplate +from liquid import CachingFileSystemLoader +from liquid import Context +from liquid import Environment + + +class CachingFileSystemLoaderTestCase(unittest.TestCase): + def test_load_template(self): + """Test that we can load a template from the file system.""" + env = Environment(loader=CachingFileSystemLoader(search_path="tests/fixtures/")) + template = env.get_template(name="dropify/index.liquid") + self.assertIsInstance(template, BoundTemplate) + + def test_load_template_async(self): + """Test that we can load a template from the file system asynchronously.""" + env = Environment(loader=CachingFileSystemLoader(search_path="tests/fixtures/")) + + async def coro() -> BoundTemplate: + return await env.get_template_async(name="dropify/index.liquid") + + template = asyncio.run(coro()) + self.assertIsInstance(template, BoundTemplate) + + def test_cached_template(self): + """Test that templates loaded from the file system get cached.""" + loader = CachingFileSystemLoader(search_path="tests/fixtures/") + env = Environment(loader=loader) + self.assertIsNone(env.cache) + template = env.get_template(name="dropify/index.liquid") + self.assertTrue(template.is_up_to_date) + another = env.get_template(name="dropify/index.liquid", globals={"foo": "bar"}) + self.assertTrue(another.is_up_to_date) + self.assertIs(template.tree, another.tree) + self.assertEqual(len(loader.cache), 1) + + def test_cached_template_async(self): + """Test that async loaded templates get cached.""" + loader = CachingFileSystemLoader(search_path="tests/fixtures/") + env = Environment(loader=loader) + self.assertIsNone(env.cache) + + async def get_template() -> BoundTemplate: + return await env.get_template_async( + name="dropify/index.liquid", globals={"foo": "bar"} + ) + + async def is_up_to_date(template: BoundTemplate) -> bool: + return await template.is_up_to_date_async() + + template = asyncio.run(get_template()) + self.assertTrue(asyncio.run(is_up_to_date(template))) + another = asyncio.run(get_template()) + self.assertTrue(asyncio.run(is_up_to_date(another))) + self.assertIs(template.tree, another.tree) + self.assertEqual(len(loader.cache), 1) + + def test_auto_reload_template(self): + """Test templates loaded from the file system are reloaded automatically.""" + with tempfile.TemporaryDirectory() as tmpdir: + template_path = Path(tmpdir) / "somefile.txt" + + # Initial template content + with template_path.open("w", encoding="UTF-8") as fd: + fd.write("hello there\n") + + env = Environment(loader=CachingFileSystemLoader(search_path=tmpdir)) + self.assertIsNone(env.cache) + + async def get_template() -> BoundTemplate: + return await env.get_template_async(name=str(template_path)) + + async def is_up_to_date(template: BoundTemplate) -> bool: + return await template.is_up_to_date_async() + + template = asyncio.run(get_template()) + self.assertTrue(asyncio.run(is_up_to_date(template))) + + same_template = asyncio.run(get_template()) + self.assertTrue(asyncio.run(is_up_to_date(template))) + self.assertIs(template.tree, same_template.tree) + + # Update template content. + time.sleep(0.01) # Make sure some time has passed. + template_path.touch() + + # Template has been updated + self.assertFalse(asyncio.run(is_up_to_date(template))) + updated_template = asyncio.run(get_template()) + self.assertTrue(asyncio.run(is_up_to_date(updated_template))) + self.assertIsNot(template.tree, updated_template.tree) + + def test_auto_reload_template_async(self): + """Test templates loaded from the file system are reloaded automatically.""" + with tempfile.TemporaryDirectory() as tmpdir: + template_path = Path(tmpdir) / "somefile.txt" + + # Initial template content + with template_path.open("w", encoding="UTF-8") as fd: + fd.write("hello there\n") + + env = Environment(loader=CachingFileSystemLoader(search_path=tmpdir)) + self.assertIsNone(env.cache) + + template = env.get_template(name=str(template_path)) + self.assertTrue(template.is_up_to_date) + + same_template = env.get_template(name=str(template_path)) + self.assertTrue(template.is_up_to_date) + self.assertEqual(template.tree, same_template.tree) + + # Update template content. + time.sleep(0.01) # Make sure some time has passed. + template_path.touch() + + # Template has been updated + self.assertFalse(template.is_up_to_date) + updated_template = env.get_template(name=str(template_path)) + self.assertTrue(updated_template.is_up_to_date) + self.assertIsNot(template.tree, updated_template.tree) + + def test_without_auto_reload_template(self): + """Test that auto_reload can be disabled.""" + with tempfile.TemporaryDirectory() as tmpdir: + template_path = Path(tmpdir) / "somefile.txt" + + # Initial template content + with template_path.open("w", encoding="UTF-8") as fd: + fd.write("hello there\n") + + env = Environment( + loader=CachingFileSystemLoader(search_path=tmpdir, auto_reload=False) + ) + self.assertIsNone(env.cache) + + template = env.get_template(name=str(template_path)) + self.assertTrue(template.is_up_to_date) + + same_template = env.get_template(name=str(template_path)) + self.assertTrue(template.is_up_to_date) + self.assertEqual(template.tree, same_template.tree) + + # Update template content. + time.sleep(0.01) # Make sure some time has passed. + template_path.touch() + + # Template has been updated + self.assertFalse(template.is_up_to_date) + updated_template = env.get_template(name=str(template_path)) + self.assertFalse(updated_template.is_up_to_date) + self.assertEqual(template.tree, updated_template.tree) + + def test_load_with_args(self): + """Test that we default to an empty namespace, ignoring extra args.""" + loader = CachingFileSystemLoader(search_path="tests/fixtures/") + env = Environment(loader=loader) + self.assertIsNone(env.cache) + + template = env.get_template_with_args(name="dropify/index.liquid", foo="bar") + self.assertIsInstance(template, BoundTemplate) + + _template = asyncio.run( + env.get_template_with_args_async(name="dropify/index.liquid", foo="bar") + ) + self.assertIsInstance(_template, BoundTemplate) + self.assertIs(_template, template) + + def test_load_from_namespace_with_args(self): + """Test that we can provide a namespace with args.""" + loader = CachingFileSystemLoader( + search_path="tests/fixtures/", namespace_key="foo" + ) + env = Environment(loader=loader) + self.assertIsNone(env.cache) + + template = env.get_template_with_args(name="dropify/index.liquid") + self.assertIsInstance(template, BoundTemplate) + + _template = asyncio.run( + env.get_template_with_args_async( + name="dropify/index.liquid", foo="namespace" + ) + ) + self.assertIsInstance(_template, BoundTemplate) + self.assertIsNot(_template, template) + self.assertEqual(_template.render(), "namespaced template") + + def test_load_with_context(self): + """Test that we can load a cached template referencing a render context.""" + loader = CachingFileSystemLoader( + search_path="tests/fixtures/", namespace_key="foo" + ) + env = Environment(loader=loader) + self.assertIsNone(env.cache) + context = Context(env=env, globals={"foo": "namespace"}) + template = env.get_template_with_context(context, "dropify/index.liquid") + self.assertEqual(template.render(), "namespaced template") + + def test_load_with_context_async(self): + """Test that we can load a cached template referencing a render context.""" + loader = CachingFileSystemLoader( + search_path="tests/fixtures/", namespace_key="foo" + ) + env = Environment(loader=loader) + self.assertIsNone(env.cache) + context = Context(env=env, globals={"foo": "namespace"}) + + async def coro() -> BoundTemplate: + return await env.get_template_with_context_async( + context, "dropify/index.liquid" + ) + + self.assertEqual(asyncio.run(coro()).render(), "namespaced template") + + def test_load_with_context_no_namespace(self): + """Test that we can load a cached template referencing a render context.""" + loader = CachingFileSystemLoader(search_path="tests/fixtures/") + env = Environment(loader=loader) + self.assertIsNone(env.cache) + context = Context(env=env, globals={"foo": "namespace"}) + template = env.get_template_with_context(context, "dropify/index.liquid") + self.assertNotEqual(template.render(), "namespaced template") + + def test_load_with_context_missing_namespace(self): + """Test that we fall back to an unscoped template name.""" + loader = CachingFileSystemLoader( + search_path="tests/fixtures/", namespace_key="foo" + ) + env = Environment(loader=loader) + self.assertIsNone(env.cache) + context = Context(env=env) + template = env.get_template_with_context(context, "dropify/index.liquid") + self.assertNotEqual(template.render(), "namespaced template")