Skip to content

jinjaloaderfilesystem

Class info

Classes

Name Children Inherits
JinjaLoaderFileSystem
jinjarope.jinjaloaderfilesystem
A FsSpec Filesystem implementation for jinja environment templates.

    🛈 DocStrings

    Filesystem implementation for jinja environment templates.

    JinjaLoaderFileSystem

    Bases: AbstractFileSystem

    A FsSpec Filesystem implementation for jinja environment templates.

    Source code in src/jinjarope/jinjaloaderfilesystem.py 
     17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
    class JinjaLoaderFileSystem(fsspec.AbstractFileSystem):
    """A FsSpec Filesystem implementation for jinja environment templates."""

    protocol = "jinja"
    async_impl = True

    def __init__(self, env: jinja2.Environment) -> None:
        """Initialize a JinjaLoader filesystem.

        Args:
            env: The environment of the loaders to get a filesystem for.
        """
        super().__init__()
        self.env = env

    @override
    def isdir(self, path: str) -> bool:
        """Check if path is a directory.

        Args:
            path: Path to check

        Returns:
            True if path is a directory
        """
        if not self.env.loader:
            return False

        clean_path = UPath(path).as_posix().strip("/")
        if clean_path in {"", "/", "."}:
            return True

        templates = self.env.loader.list_templates()
        return any(template.startswith(f"{clean_path}/") for template in templates)

    @override
    def isfile(self, path: str) -> bool:
        """Check if path is a file.

        Args:
            path: Path to check

        Returns:
            True if path is a file
        """
        if not self.env.loader:
            return False

        try:
            self.env.loader.get_source(self.env, path)
        except jinja2.TemplateNotFound:
            return False
        else:
            return True

    @override
    def cat_file(self, path: str, **kwargs: Any) -> bytes:
        """Get contents of file as bytes.

        Args:
            path: Path to file
            **kwargs: Additional arguments (unused)

        Returns:
            File contents as bytes

        Raises:
            FileNotFoundError: If template not found or env has no loader
        """
        if not self.env.loader:
            msg = "Environment has no loader"
            raise FileNotFoundError(msg)

        try:
            source, _, _ = self.env.loader.get_source(self.env, path)
            return source.encode()
        except jinja2.TemplateNotFound as exc:
            msg = f"Template not found: {path}"
            raise FileNotFoundError(msg) from exc

    @override
    def cat(self, path: str | list[str], **kwargs: Any) -> bytes | dict[str, bytes]:
        """Get contents of file(s).

        Args:
            path: Path or list of paths
            **kwargs: Additional arguments

        Returns:
            File contents as bytes or dict of path -> contents
        """
        if isinstance(path, str):
            return self.cat_file(path, **kwargs)

        return {p: self.cat_file(p, **kwargs) for p in path}

    async def _cat_file(self, path: str, **kwargs: Any) -> bytes:
        """Async implementation of cat_file.

        Args:
            path: Path to file
            **kwargs: Additional arguments

        Returns:
            File contents as bytes
        """
        return self.cat_file(path, **kwargs)

    @override
    def ls(
        self, path: str, detail: bool = True, **kwargs: Any
    ) -> list[dict[str, str]] | list[str]:
        """List contents of path.

        Args:
            path: The path to list
            detail: If True, return a list of dictionaries, else return a list of paths
            **kwargs: Additional arguments (unused)

        Returns:
            List of paths or file details

        Raises:
            FileNotFoundError: If path doesn't exist or env has no loader
        """
        if not self.env.loader:
            msg = "Environment has no loader"
            raise FileNotFoundError(msg)

        templates = self.env.loader.list_templates()
        clean_path = UPath(path).as_posix().strip("/")

        if clean_path in {"", "/", "."}:
            return self._list_root(templates, detail)
        return self._list_subdirectory(templates, clean_path, detail)

    def _list_root(
        self, templates: list[str], detail: bool
    ) -> list[dict[str, str]] | list[str]:
        """List contents of root directory."""
        root_files = [path for path in templates if "/" not in path]
        root_dirs = {
            path.split("/")[0]
            for path in templates
            if "/" in path and path not in root_files
        }

        if detail:
            file_entries = [{"name": path, "type": "file"} for path in root_files]
            dir_entries = [{"name": path, "type": "directory"} for path in root_dirs]
            return dir_entries + file_entries
        return list(root_dirs) + root_files

    def _list_subdirectory(
        self, templates: list[str], path: str, detail: bool
    ) -> list[dict[str, str]] | list[str]:
        """List contents of a subdirectory."""
        # Get all templates that start with the path
        relevant_templates = [
            template for template in templates if template.startswith(f"{path}/")
        ]

        if not relevant_templates:
            msg = f"Directory not found: {path}"
            raise FileNotFoundError(msg)

        # Get immediate children only
        items: set[str] = set()
        for template in relevant_templates:
            # Remove the path prefix and split the remaining path
            relative_path = template[len(f"{path}/") :].split("/", 1)[0]
            items.add(relative_path)

        # Sort for consistent ordering
        sorted_items = sorted(items)

        if detail:
            return [
                {
                    "name": item,
                    # If there's no extension or if it appears in the full paths
                    # with something after it, it's a directory
                    "type": "directory"
                    if (
                        "." not in item
                        or any(t.startswith(f"{path}/{item}/") for t in templates)
                    )
                    else "file",
                }
                for item in sorted_items
            ]
        return list(sorted_items)

    async def _ls(
        self, path: str, detail: bool = True, **kwargs: Any
    ) -> list[dict[str, str]] | list[str]:
        """Async implementation of ls."""
        return self.ls(path, detail=detail, **kwargs)

    @override
    def _open(
        self,
        path: str,
        mode: Literal["rb", "wb", "ab"] = "rb",
        block_size: int | None = None,
        autocommit: bool = True,
        cache_options: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> memory.MemoryFile:
        """Open a file."""
        if not self.env.loader:
            msg = "Environment has no loader"
            raise FileNotFoundError(msg)

        try:
            source, _, _ = self.env.loader.get_source(self.env, path)
            return memory.MemoryFile(fs=self, path=path, data=source.encode())
        except jinja2.TemplateNotFound as exc:
            msg = f"Template not found: {path}"
            raise FileNotFoundError(msg) from exc

    async def _open_async(
        self,
        path: str,
        mode: Literal["rb", "wb", "ab"] = "rb",
        block_size: int | None = None,
        autocommit: bool = True,
        cache_options: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> memory.MemoryFile:
        """Async implementation of _open."""
        return self._open(
            path,
            mode=mode,
            block_size=block_size,
            autocommit=autocommit,
            cache_options=cache_options,
            **kwargs,
        )

    @override
    def exists(self, path: str, **_kwargs: Any) -> bool:
        """Check if path exists.

        Args:
            path: Path to check

        Returns:
            True if path exists as a file or directory
        """
        return self.isfile(path) or self.isdir(path)

    @override
    def info(self, path: str, **kwargs: Any) -> dict[str, Any]:
        """Get info about a path.

        Args:
            path: Path to get info for
            **kwargs: Additional arguments (unused)

        Returns:
            Dictionary of path information

        Raises:
            FileNotFoundError: If path doesn't exist
        """
        if self.isfile(path):
            content = self.cat_file(path)
            return {
                "name": path,
                "size": len(content),
                "type": "file",
                "created": None,  # Jinja doesn't track these
                "modified": None,
            }
        if self.isdir(path):
            return {
                "name": path,
                "size": 0,
                "type": "directory",
                "created": None,
                "modified": None,
            }

        msg = f"Path not found: {path}"
        raise FileNotFoundError(msg)

    __init__ 

    __init__(env: Environment) -> None

    Initialize a JinjaLoader filesystem.

    Parameters:

    Name Type Description Default
    env Environment

    The environment of the loaders to get a filesystem for.

    		 required
    Source code in src/jinjarope/jinjaloaderfilesystem.py 
    23
24
25
26
27
28
29
30
    def __init__(self, env: jinja2.Environment) -> None:
    """Initialize a JinjaLoader filesystem.

    Args:
        env: The environment of the loaders to get a filesystem for.
    """
    super().__init__()
    self.env = env

    cat 

    cat(path: str | list[str], **kwargs: Any) -> bytes | dict[str, bytes]

    Get contents of file(s).

    Parameters:

    Name Type Description Default
    path str | list[str]

    Path or list of paths

    		 required
    **kwargs Any

    Additional arguments

    		 {}

    Returns:

    Type Description
    bytes | dict[str, bytes]

    File contents as bytes or dict of path -> contents
    Source code in src/jinjarope/jinjaloaderfilesystem.py 
     97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
    @override
def cat(self, path: str | list[str], **kwargs: Any) -> bytes | dict[str, bytes]:
    """Get contents of file(s).

    Args:
        path: Path or list of paths
        **kwargs: Additional arguments

    Returns:
        File contents as bytes or dict of path -> contents
    """
    if isinstance(path, str):
        return self.cat_file(path, **kwargs)

    return {p: self.cat_file(p, **kwargs) for p in path}

    cat_file 

    cat_file(path: str, **kwargs: Any) -> bytes

    Get contents of file as bytes.

    Parameters:

    Name Type Description Default
    path str

    Path to file

    		 required
    **kwargs Any

    Additional arguments (unused)

    		 {}

    Returns:

    Type Description
    bytes

    File contents as bytes

    Raises:

    Type Description
    FileNotFoundError

    If template not found or env has no loader
    Source code in src/jinjarope/jinjaloaderfilesystem.py 
    72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
    @override
def cat_file(self, path: str, **kwargs: Any) -> bytes:
    """Get contents of file as bytes.

    Args:
        path: Path to file
        **kwargs: Additional arguments (unused)

    Returns:
        File contents as bytes

    Raises:
        FileNotFoundError: If template not found or env has no loader
    """
    if not self.env.loader:
        msg = "Environment has no loader"
        raise FileNotFoundError(msg)

    try:
        source, _, _ = self.env.loader.get_source(self.env, path)
        return source.encode()
    except jinja2.TemplateNotFound as exc:
        msg = f"Template not found: {path}"
        raise FileNotFoundError(msg) from exc

    exists 

    exists(path: str, **_kwargs: Any) -> bool

    Check if path exists.

    Parameters:

    Name Type Description Default
    path str

    Path to check

    		 required

    Returns:

    Type Description
    bool

    True if path exists as a file or directory
    Source code in src/jinjarope/jinjaloaderfilesystem.py 
    257
258
259
260
261
262
263
264
265
266
267
    @override
def exists(self, path: str, **_kwargs: Any) -> bool:
    """Check if path exists.

    Args:
        path: Path to check

    Returns:
        True if path exists as a file or directory
    """
    return self.isfile(path) or self.isdir(path)

    info 

    info(path: str, **kwargs: Any) -> dict[str, Any]

    Get info about a path.

    Parameters:

    Name Type Description Default
    path str

    Path to get info for

    		 required
    **kwargs Any

    Additional arguments (unused)

    		 {}

    Returns:

    Type Description
    dict[str, Any]

    Dictionary of path information

    Raises:

    Type Description
    FileNotFoundError

    If path doesn't exist
    Source code in src/jinjarope/jinjaloaderfilesystem.py 
    269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
    @override
def info(self, path: str, **kwargs: Any) -> dict[str, Any]:
    """Get info about a path.

    Args:
        path: Path to get info for
        **kwargs: Additional arguments (unused)

    Returns:
        Dictionary of path information

    Raises:
        FileNotFoundError: If path doesn't exist
    """
    if self.isfile(path):
        content = self.cat_file(path)
        return {
            "name": path,
            "size": len(content),
            "type": "file",
            "created": None,  # Jinja doesn't track these
            "modified": None,
        }
    if self.isdir(path):
        return {
            "name": path,
            "size": 0,
            "type": "directory",
            "created": None,
            "modified": None,
        }

    msg = f"Path not found: {path}"
    raise FileNotFoundError(msg)

    isdir 

    isdir(path: str) -> bool

    Check if path is a directory.

    Parameters:

    Name Type Description Default
    path str

    Path to check

    		 required

    Returns:

    Type Description
    bool

    True if path is a directory
    Source code in src/jinjarope/jinjaloaderfilesystem.py 
    32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
    @override
def isdir(self, path: str) -> bool:
    """Check if path is a directory.

    Args:
        path: Path to check

    Returns:
        True if path is a directory
    """
    if not self.env.loader:
        return False

    clean_path = UPath(path).as_posix().strip("/")
    if clean_path in {"", "/", "."}:
        return True

    templates = self.env.loader.list_templates()
    return any(template.startswith(f"{clean_path}/") for template in templates)

    isfile 

    isfile(path: str) -> bool

    Check if path is a file.

    Parameters:

    Name Type Description Default
    path str

    Path to check

    		 required

    Returns:

    Type Description
    bool

    True if path is a file
    Source code in src/jinjarope/jinjaloaderfilesystem.py 
    52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
    @override
def isfile(self, path: str) -> bool:
    """Check if path is a file.

    Args:
        path: Path to check

    Returns:
        True if path is a file
    """
    if not self.env.loader:
        return False

    try:
        self.env.loader.get_source(self.env, path)
    except jinja2.TemplateNotFound:
        return False
    else:
        return True

    ls 

    ls(path: str, detail: bool = True, **kwargs: Any) -> list[dict[str, str]] | list[str]

    List contents of path.

    Parameters:

    Name Type Description Default
    path str

    The path to list

    		 required
    detail bool

    If True, return a list of dictionaries, else return a list of paths

    		 True
    **kwargs Any

    Additional arguments (unused)

    		 {}

    Returns:

    Type Description
    list[dict[str, str]] | list[str]

    List of paths or file details

    Raises:

    Type Description
    FileNotFoundError

    If path doesn't exist or env has no loader
    Source code in src/jinjarope/jinjaloaderfilesystem.py 
    125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
    @override
def ls(
    self, path: str, detail: bool = True, **kwargs: Any
) -> list[dict[str, str]] | list[str]:
    """List contents of path.

    Args:
        path: The path to list
        detail: If True, return a list of dictionaries, else return a list of paths
        **kwargs: Additional arguments (unused)

    Returns:
        List of paths or file details

    Raises:
        FileNotFoundError: If path doesn't exist or env has no loader
    """
    if not self.env.loader:
        msg = "Environment has no loader"
        raise FileNotFoundError(msg)

    templates = self.env.loader.list_templates()
    clean_path = UPath(path).as_posix().strip("/")

    if clean_path in {"", "/", "."}:
        return self._list_root(templates, detail)
    return self._list_subdirectory(templates, clean_path, detail)