FsSpecFileSystemLoader

Base classes¶

Name	Children	Inherits
LoaderMixin jinjarope.loaders Loader mixin which allows to OR loaders into a choice loader.	PrefixLoader ModuleLoader FunctionLoader PackageLoader FileSystemLoader ChoiceLoader DictLoader RewriteLoader NestedDictLoader FsSpecProtocolPathLoader FsSpecFileSystemLoader ...	object
BaseLoader jinja2.loaders Baseclass for all loaders. Subclass this and override `get_source` to	FileSystemLoader PackageLoader DictLoader FunctionLoader PrefixLoader ChoiceLoader ModuleLoader RewriteLoader NestedDictLoader FsSpecProtocolPathLoader FsSpecFileSystemLoader ...	object

⋔ Inheritance diagram¶

graph TD
  94892546070672["fsspecloaders.FsSpecFileSystemLoader"]
  94892548325824["loaders.LoaderMixin"]
  140247609577664["builtins.object"]
  94892544482432["loaders.BaseLoader"]
  94892548325824 --> 94892546070672
  140247609577664 --> 94892548325824
  94892544482432 --> 94892546070672
  140247609577664 --> 94892544482432

🛈 DocStrings¶

Bases: LoaderMixin, BaseLoader

A jinja loader for fsspec filesystems.

This loader allows to access templates from an fsspec filesystem.

Template paths must be relative to the filesystem root. In order to access templates via protocol path, see FsSpecProtocolPathLoader.

Examples:

# protocol path
loader = FsSpecFileSystemLoader("dir::github://phil65:mknodes@main/docs")
env = Environment(loader=loader)
env.get_template("icons.jinja").render()

# protocol and storage options
loader = FsSpecFileSystemLoader("github", org="phil65", repo="mknodes")
env = Environment(loader=loader)
env.get_template("docs/icons.jinja").render()

# fsspec filesystem
fs = fsspec.filesystem("github", org="phil65", repo="mknodes")
loader = FsSpecFileSystemLoader(fs)
env = Environment(loader=loader)
env.get_template("docs/icons.jinja").render()

Source code in src/jinjarope/fsspecloaders.py

class FsSpecFileSystemLoader(loaders_.LoaderMixin, jinja2.BaseLoader):
    """A jinja loader for fsspec filesystems.

    This loader allows to access templates from an fsspec filesystem.

    Template paths must be relative to the filesystem root.
    In order to access templates via protocol path, see `FsSpecProtocolPathLoader`.

    Examples:
        ``` py
        # protocol path
        loader = FsSpecFileSystemLoader("dir::github://phil65:mknodes@main/docs")
        env = Environment(loader=loader)
        env.get_template("icons.jinja").render()

        # protocol and storage options
        loader = FsSpecFileSystemLoader("github", org="phil65", repo="mknodes")
        env = Environment(loader=loader)
        env.get_template("docs/icons.jinja").render()

        # fsspec filesystem
        fs = fsspec.filesystem("github", org="phil65", repo="mknodes")
        loader = FsSpecFileSystemLoader(fs)
        env = Environment(loader=loader)
        env.get_template("docs/icons.jinja").render()
        ```

    """

    ID = "fsspec"

    def __init__(self, fs: fsspec.AbstractFileSystem | str, **kwargs: Any):
        """Constructor.

        Args:
            fs: Either a protocol path string or an fsspec filesystem instance.
                Also supports "::dir" prefix to set the root path.
            kwargs: Optional storage options for the filesystem.
        """
        import fsspec
        import fsspec.core

        super().__init__()
        match fs:
            case str() if "://" in fs:
                self.fs, self.path = fsspec.core.url_to_fs(fs, **kwargs)
            case str():
                self.fs, self.path = fsspec.filesystem(fs, **kwargs), ""
            case _:
                self.fs, self.path = fs, ""
        self.storage_options = kwargs

    def __repr__(self):
        return utils.get_repr(self, fs=self.fs)

    def __eq__(self, other):
        return (
            type(self) is type(other)
            and self.storage_options == other.storage_options
            and self.fs == other.fs
            and self.path == other.path
        )

    def __hash__(self):
        return (
            hash(tuple(sorted(self.storage_options.items())))
            + hash(self.fs)
            + hash(self.path)
        )

    def list_templates(self) -> list[str]:
        return [
            f"{path}{self.fs.sep}{f}" if path else f
            for path, _dirs, files in self.fs.walk(self.fs.root_marker)
            for f in files
        ]

    def get_source(
        self,
        environment: jinja2.Environment,
        template: str,
    ) -> tuple[str, str, Callable[[], bool] | None]:
        try:
            with self.fs.open(template) as file:
                src = file.read().decode()  # pyright: ignore
        except FileNotFoundError as e:
            raise jinja2.TemplateNotFound(template) from e
        path = pathlib.Path(template).as_posix()
        return src, path, lambda: True

    async def get_source_async(
        self,
        environment: jinja2.Environment,
        template: str,
    ) -> tuple[str, str, Callable[[], bool] | None]:
        from upathtools.async_ops import get_async_fs

        fs = await get_async_fs(self.fs)
        try:
            file = await fs.open_async(template)
            async with file:
                src = await file.read().decode()  # pyright: ignore
        except FileNotFoundError as e:
            raise jinja2.TemplateNotFound(template) from e
        path = pathlib.Path(template).as_posix()
        return src, path, lambda: True

init ¶

__init__(fs: AbstractFileSystem | str, **kwargs: Any)

Constructor.

Parameters:

Name	Type	Description	Default
`fs`	`AbstractFileSystem \| str`	Either a protocol path string or an fsspec filesystem instance. Also supports "::dir" prefix to set the root path.	required
`kwargs`	`Any`	Optional storage options for the filesystem.	`{}`