1# SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2# SPDX-License-Identifier: Apache-2.0
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15
16import glob
17import json
18import logging
19import os
20import shutil
21import tempfile
22from collections.abc import Callable, Iterator
23from datetime import datetime, timezone
24from enum import Enum
25from io import BytesIO, StringIO
26from typing import IO, Any, TypeVar
27
28import xattr
29
30from .. import _xattr
31from ..telemetry import Telemetry
32from ..types import AWARE_DATETIME_MIN, ObjectMetadata, Range, SymlinkHandling
33from ..utils import (
34 create_attribute_filter_evaluator,
35 matches_attribute_filter_expression,
36 safe_makedirs,
37 validate_attributes,
38)
39from .base import BaseStorageProvider
40
41_T = TypeVar("_T")
42
43PROVIDER = "file"
44READ_CHUNK_SIZE = 8192
45
46logger = logging.getLogger(__name__)
47
48
49class _EntryType(Enum):
50 """
51 An enum representing the type of an entry in a directory.
52 """
53
54 FILE = 1
55 DIRECTORY = 2
56 DIRECTORY_TO_EXPLORE = 3
57 SYMLINK = 4
58
59
[docs]
60def atomic_write(source: str | IO, destination: str, attributes: dict[str, str] | None = None):
61 """
62 Writes the contents of a file to the specified destination path.
63
64 This function ensures that the file write operation is atomic, meaning the output file is either fully written or not modified at all.
65 This is achieved by writing to a temporary file first and then renaming it to the destination path.
66
67 :param source: The input file to read from. It can be a string representing the path to a file, or an open file-like object (IO).
68 :param destination: The path to the destination file where the contents should be written.
69 :param attributes: The attributes to set on the file.
70 """
71
72 with tempfile.NamedTemporaryFile(mode="wb", delete=False, dir=os.path.dirname(destination), prefix=".") as fp:
73 temp_file_path = fp.name
74 if isinstance(source, str):
75 with open(source, mode="rb") as src:
76 while chunk := src.read(READ_CHUNK_SIZE):
77 fp.write(chunk)
78 else:
79 while chunk := source.read(READ_CHUNK_SIZE):
80 fp.write(chunk)
81
82 # Set attributes on temp file if provided
83 validated_attributes = validate_attributes(attributes)
84 if validated_attributes:
85 try:
86 xattr.setxattr(temp_file_path, "user.json", json.dumps(validated_attributes).encode("utf-8"))
87 except OSError as e:
88 logger.debug(f"Failed to set extended attributes on temp file {temp_file_path}: {e}")
89
90 os.rename(src=temp_file_path, dst=destination)
91
92
[docs]
93class PosixFileStorageProvider(BaseStorageProvider):
94 """
95 A concrete implementation of the :py:class:`multistorageclient.types.StorageProvider` for interacting with POSIX file systems.
96 """
97
98 def __init__(
99 self,
100 base_path: str,
101 config_dict: dict[str, Any] | None = None,
102 telemetry_provider: Callable[[], Telemetry] | None = None,
103 **kwargs: Any,
104 ) -> None:
105 """
106 :param base_path: The root prefix path within the POSIX file system where all operations will be scoped.
107 :param config_dict: Resolved MSC config.
108 :param telemetry_provider: A function that provides a telemetry instance.
109 """
110
111 # Validate POSIX path
112 if base_path == "":
113 base_path = "/"
114
115 if not base_path.startswith("/"):
116 raise ValueError(f"The base_path {base_path} must be an absolute path.")
117
118 super().__init__(
119 base_path=base_path,
120 provider_name=PROVIDER,
121 config_dict=config_dict,
122 telemetry_provider=telemetry_provider,
123 )
124 self._real_base_path = os.path.realpath(base_path)
125
126 def _translate_errors(
127 self,
128 func: Callable[[], _T],
129 operation: str,
130 path: str,
131 ) -> _T:
132 """
133 Translates errors like timeouts and client errors.
134
135 :param func: The function that performs the actual file operation.
136 :param operation: The type of operation being performed (e.g., "PUT", "GET", "DELETE").
137 :param path: The path to the object.
138
139 :return: The result of the file operation, typically the return value of the `func` callable.
140 """
141 try:
142 return func()
143 except FileNotFoundError:
144 raise
145 except Exception as error:
146 raise RuntimeError(f"Failed to {operation} object(s) at {path}, error: {error}") from error
147
148 def _put_object(
149 self,
150 path: str,
151 body: bytes,
152 if_match: str | None = None,
153 if_none_match: str | None = None,
154 attributes: dict[str, str] | None = None,
155 ) -> int:
156 def _invoke_api() -> int:
157 safe_makedirs(os.path.dirname(path))
158 atomic_write(source=BytesIO(body), destination=path, attributes=attributes)
159 return len(body)
160
161 return self._translate_errors(_invoke_api, operation="PUT", path=path)
162
163 def _get_object(self, path: str, byte_range: Range | None = None) -> bytes:
164 def _invoke_api() -> bytes:
165 if byte_range:
166 with open(path, "rb") as f:
167 f.seek(byte_range.offset)
168 return f.read(byte_range.size)
169 else:
170 with open(path, "rb") as f:
171 return f.read()
172
173 return self._translate_errors(_invoke_api, operation="GET", path=path)
174
175 def _copy_object(self, src_path: str, dest_path: str) -> int:
176 src_object = self._get_object_metadata(src_path)
177
178 def _invoke_api() -> int:
179 safe_makedirs(os.path.dirname(dest_path))
180 atomic_write(source=src_path, destination=dest_path, attributes=src_object.metadata)
181
182 return src_object.content_length
183
184 return self._translate_errors(_invoke_api, operation="COPY", path=src_path)
185
186 def _delete_object(self, path: str, if_match: str | None = None) -> None:
187 def _invoke_api() -> None:
188 if os.path.exists(path) and os.path.isfile(path):
189 os.remove(path)
190
191 return self._translate_errors(_invoke_api, operation="DELETE", path=path)
192
193 def _make_symlink(self, path: str, target: str) -> None:
194 def _invoke_api() -> None:
195 safe_makedirs(os.path.dirname(path))
196 relative_target = ObjectMetadata.encode_symlink_target(path, target)
197 if os.path.lexists(path):
198 os.remove(path)
199 os.symlink(relative_target, path)
200
201 self._translate_errors(_invoke_api, operation="SYMLINK", path=path)
202
203 def _get_object_metadata(self, path: str, strict: bool = True) -> ObjectMetadata:
204 is_dir = os.path.isdir(path)
205 if is_dir:
206 path = self._append_delimiter(path)
207
208 def _invoke_api() -> ObjectMetadata:
209 metadata_dict = {}
210 try:
211 json_bytes = _xattr.getxattr(path, "user.json")
212 metadata_dict = json.loads(json_bytes.decode("utf-8"))
213 except (OSError, KeyError, json.JSONDecodeError, AttributeError) as e:
214 logger.debug(f"Failed to read extended attributes from {path}: {e}")
215
216 # ``os.readlink`` may return an absolute path; normalise to the
217 # parent-relative form used by every backend.
218 symlink_target: str | None = None
219 if os.path.islink(path):
220 raw = os.readlink(path)
221 symlink_target = ObjectMetadata.encode_symlink_target(path, raw) if os.path.isabs(raw) else raw
222
223 return ObjectMetadata(
224 key=path,
225 type="directory" if is_dir else "file",
226 content_length=0 if is_dir else os.path.getsize(path),
227 last_modified=datetime.fromtimestamp(os.path.getmtime(path), tz=timezone.utc),
228 metadata=metadata_dict if metadata_dict else None,
229 symlink_target=symlink_target,
230 )
231
232 return self._translate_errors(_invoke_api, operation="HEAD", path=path)
233
234 def _list_objects(
235 self,
236 path: str,
237 start_after: str | None = None,
238 end_at: str | None = None,
239 include_directories: bool = False,
240 symlink_handling: SymlinkHandling = SymlinkHandling.FOLLOW,
241 ) -> Iterator[ObjectMetadata]:
242 start_after = os.path.relpath(start_after, self._base_path) if start_after else None
243 end_at = os.path.relpath(end_at, self._base_path) if end_at else None
244
245 def _invoke_api() -> Iterator[ObjectMetadata]:
246 symlink_path = path.rstrip("/") or path
247 if os.path.islink(symlink_path):
248 if symlink_handling == SymlinkHandling.SKIP:
249 return
250
251 relative_path = self._validate_symlink_target(symlink_path, symlink_handling)
252 if relative_path is None:
253 return
254
255 if (start_after is not None and relative_path <= start_after) or (
256 end_at is not None and relative_path > end_at
257 ):
258 return
259
260 if symlink_handling in (SymlinkHandling.PRESERVE, SymlinkHandling.PRESERVE_STRICT):
261 relative_target, target_type = self._get_symlink_target_info(symlink_path)
262 yield ObjectMetadata(
263 key=relative_path,
264 content_length=0,
265 last_modified=datetime.fromtimestamp(os.path.getmtime(symlink_path), tz=timezone.utc),
266 type=target_type,
267 symlink_target=relative_target,
268 )
269 return
270
271 if os.path.isfile(path):
272 relative_path = os.path.relpath(path, self._base_path)
273 if (start_after is None or relative_path > start_after) and (end_at is None or relative_path <= end_at):
274 yield ObjectMetadata(
275 key=relative_path,
276 content_length=os.path.getsize(path),
277 last_modified=datetime.fromtimestamp(os.path.getmtime(path), tz=timezone.utc),
278 )
279 dir_path = path.rstrip("/") + "/"
280 if not os.path.isdir(dir_path):
281 return
282
283 yield from self._explore_directory(dir_path, start_after, end_at, include_directories, symlink_handling)
284
285 return self._translate_errors(_invoke_api, operation="LIST", path=path)
286
287 @property
288 def supports_parallel_listing(self) -> bool:
289 """
290 Whether this provider supports heap-based parallel recursive listing.
291
292 :return: ``True`` for POSIX file storage.
293 """
294 return True
295
296 def _shallow_list(self, path: str, symlink_handling: SymlinkHandling) -> tuple[list[str], list[ObjectMetadata]]:
297 """
298 Adapt POSIX relative listing keys to the full-key contract expected by the recursive listing heap.
299 """
300 prefixes: list[str] = []
301 objects: list[ObjectMetadata] = []
302
303 for item in self._list_objects(path, include_directories=True, symlink_handling=symlink_handling):
304 full_key = self._prepend_base_path(item.key)
305 if item.type == "directory" and item.symlink_target is None:
306 child_prefix = full_key + "/"
307 if child_prefix != path.rstrip("/") + "/":
308 prefixes.append(child_prefix)
309 else:
310 item.key = full_key
311 objects.append(item)
312
313 return prefixes, objects
314
315 def _symlink_target_is_external(self, real_target: str) -> bool:
316 """
317 Return True when ``real_target`` resolves outside ``base_path``.
318
319 Uses the pre-computed ``_real_base_path`` so intermediate directory
320 symlinks do not make an in-tree target look external.
321 """
322 try:
323 return os.path.commonpath((self._real_base_path, real_target)) != self._real_base_path
324 except ValueError:
325 return True
326
327 def _validate_symlink_target(
328 self,
329 full_path: str,
330 symlink_handling: SymlinkHandling,
331 ) -> str | None:
332 """
333 Validate a symlink target and return its path relative to ``base_path``.
334
335 Non-strict modes return ``None`` for external targets, while strict
336 modes raise. Broken symlinks always raise unless the caller skipped
337 validation through ``SymlinkHandling.SKIP``.
338 """
339 relative_path = os.path.relpath(full_path, self._base_path)
340 real_target = os.path.realpath(full_path)
341 if not os.path.exists(real_target):
342 raise FileNotFoundError(
343 f"Broken symlink '{relative_path}' points to a missing target "
344 f"({full_path} -> {real_target}). Use symlink_handling=SKIP to ignore."
345 )
346
347 if not self._symlink_target_is_external(real_target):
348 return relative_path
349
350 if symlink_handling in (SymlinkHandling.FOLLOW_STRICT, SymlinkHandling.PRESERVE_STRICT):
351 raise ValueError(
352 f"Symlink '{relative_path}' points outside the base directory "
353 f"({full_path} -> {real_target}). Use symlink_handling=SKIP to ignore."
354 )
355 return None
356
357 @staticmethod
358 def _get_symlink_target_info(full_path: str) -> tuple[str, str]:
359 """Return the encoded immediate target and resolved target type."""
360 raw_link = os.readlink(full_path)
361 immediate_target = (
362 raw_link
363 if os.path.isabs(raw_link)
364 else os.path.normpath(os.path.join(os.path.dirname(full_path), raw_link))
365 )
366 relative_target = ObjectMetadata.encode_symlink_target(full_path, immediate_target)
367 target_type = "directory" if os.path.isdir(full_path) else "file"
368 return relative_target, target_type
369
370 def _explore_directory(
371 self,
372 dir_path: str,
373 start_after: str | None,
374 end_at: str | None,
375 include_directories: bool,
376 symlink_handling: SymlinkHandling = SymlinkHandling.FOLLOW,
377 ) -> Iterator[ObjectMetadata]:
378 """
379 Recursively explore a directory and yield objects in lexicographical order.
380
381 :param dir_path: The directory path to explore
382 :param start_after: The key to start after
383 :param end_at: The key to end at
384 :param include_directories: Whether to include directories in the result
385 :param symlink_handling: How to handle symbolic links during listing.
386 """
387 try:
388 dir_entries = os.listdir(dir_path)
389 dir_entries.sort()
390
391 entries: list[tuple[str, str, _EntryType]] = []
392 symlink_info: dict[str, tuple[str, str]] = {}
393
394 for entry in dir_entries:
395 full_path = os.path.join(dir_path, entry)
396 is_link = os.path.islink(full_path)
397
398 if is_link and symlink_handling == SymlinkHandling.SKIP:
399 continue
400
401 if is_link and symlink_handling in (SymlinkHandling.PRESERVE, SymlinkHandling.PRESERVE_STRICT):
402 relative_path = self._validate_symlink_target(full_path, symlink_handling)
403 if relative_path is None:
404 continue
405
406 relative_target, target_type = self._get_symlink_target_info(full_path)
407
408 if (start_after is None or start_after < relative_path) and (
409 end_at is None or relative_path <= end_at
410 ):
411 entries.append((relative_path, full_path, _EntryType.SYMLINK))
412 symlink_info[relative_path] = (relative_target, target_type)
413 continue
414
415 if is_link and symlink_handling in (SymlinkHandling.FOLLOW, SymlinkHandling.FOLLOW_STRICT):
416 relative_path = self._validate_symlink_target(full_path, symlink_handling)
417 if relative_path is None:
418 continue
419
420 relative_path = os.path.relpath(full_path, self._base_path)
421
422 if (start_after is None or start_after < relative_path) and (end_at is None or relative_path <= end_at):
423 if os.path.isfile(full_path):
424 entries.append((relative_path, full_path, _EntryType.FILE))
425 elif os.path.isdir(full_path):
426 if include_directories:
427 entries.append((relative_path, full_path, _EntryType.DIRECTORY))
428 else:
429 entries.append((relative_path, full_path, _EntryType.DIRECTORY_TO_EXPLORE))
430
431 # Sort keys must mirror the keys S3 ``list_objects_v2`` would return so POSIX
432 # listings stay in the same raw-UTF-8-byte order. For directories (expanded or
433 # returned as-is) the emitted keys live under ``<name>/``, so the trailing
434 # delimiter must be part of the sort key. Otherwise the bare name ``a`` sorts
435 # before sibling file ``a.txt`` (since ``""`` < ``".txt"``), but S3 orders the
436 # nested key ``a/b.txt`` *after* ``a.txt`` because ``.`` (0x2E) < ``/`` (0x2F).
437 def _sort_key(entry: tuple[str, str, _EntryType]) -> str:
438 relative, _, entry_type = entry
439 if entry_type in (_EntryType.DIRECTORY, _EntryType.DIRECTORY_TO_EXPLORE):
440 return relative + "/"
441 return relative
442
443 entries.sort(key=_sort_key)
444
445 for relative_path, full_path, entry_type in entries:
446 if entry_type == _EntryType.FILE:
447 yield ObjectMetadata(
448 key=relative_path,
449 content_length=os.path.getsize(full_path),
450 last_modified=datetime.fromtimestamp(os.path.getmtime(full_path), tz=timezone.utc),
451 )
452 elif entry_type == _EntryType.DIRECTORY:
453 yield ObjectMetadata(
454 key=relative_path,
455 content_length=0,
456 type="directory",
457 last_modified=AWARE_DATETIME_MIN,
458 )
459 elif entry_type == _EntryType.DIRECTORY_TO_EXPLORE:
460 yield from self._explore_directory(
461 full_path, start_after, end_at, include_directories, symlink_handling
462 )
463 elif entry_type == _EntryType.SYMLINK:
464 relative_target, target_type = symlink_info[relative_path]
465 yield ObjectMetadata(
466 key=relative_path,
467 content_length=0,
468 last_modified=datetime.fromtimestamp(os.path.getmtime(full_path), tz=timezone.utc),
469 type=target_type,
470 symlink_target=relative_target,
471 )
472
473 except FileNotFoundError:
474 raise
475 except (OSError, PermissionError) as e:
476 logger.warning(f"Failed to list contents of {dir_path}, caused by: {e}")
477 return
478
479 def _upload_file(self, remote_path: str, f: str | IO, attributes: dict[str, str] | None = None) -> int:
480 safe_makedirs(os.path.dirname(remote_path))
481
482 filesize: int = 0
483 if isinstance(f, str):
484 filesize = os.path.getsize(f)
485 elif isinstance(f, StringIO):
486 # atomic_write writes the source in binary mode; a StringIO would yield
487 # str from read() and raise TypeError on the binary write, so materialize
488 # the text as a bytes buffer here.
489 data = f.getvalue().encode("utf-8")
490 filesize = len(data)
491 f = BytesIO(data)
492 else:
493 filesize = len(f.getvalue()) # type: ignore
494
495 def _invoke_api() -> int:
496 atomic_write(source=f, destination=remote_path, attributes=attributes)
497
498 return filesize
499
500 return self._translate_errors(_invoke_api, operation="PUT", path=remote_path)
501
502 def _download_file(self, remote_path: str, f: str | IO, metadata: ObjectMetadata | None = None) -> int:
503 filesize = metadata.content_length if metadata else os.path.getsize(remote_path)
504
505 if isinstance(f, str):
506
507 def _invoke_api() -> int:
508 if os.path.dirname(f):
509 safe_makedirs(os.path.dirname(f))
510 atomic_write(source=remote_path, destination=f)
511
512 return filesize
513
514 return self._translate_errors(_invoke_api, operation="GET", path=remote_path)
515 elif isinstance(f, StringIO):
516
517 def _invoke_api() -> int:
518 with open(remote_path, "r", encoding="utf-8") as src:
519 while chunk := src.read(READ_CHUNK_SIZE):
520 f.write(chunk)
521
522 return filesize
523
524 return self._translate_errors(_invoke_api, operation="GET", path=remote_path)
525 else:
526
527 def _invoke_api() -> int:
528 with open(remote_path, "rb") as src:
529 while chunk := src.read(READ_CHUNK_SIZE):
530 f.write(chunk)
531
532 return filesize
533
534 return self._translate_errors(_invoke_api, operation="GET", path=remote_path)
535
[docs]
536 def glob(self, pattern: str, attribute_filter_expression: str | None = None) -> list[str]:
537 pattern = self._prepend_base_path(pattern)
538 keys = list(glob.glob(pattern, recursive=True))
539 if attribute_filter_expression:
540 filtered_keys = []
541 evaluator = create_attribute_filter_evaluator(attribute_filter_expression)
542 for key in keys:
543 obj_metadata = self._get_object_metadata(key)
544 if matches_attribute_filter_expression(obj_metadata, evaluator):
545 filtered_keys.append(key)
546 keys = filtered_keys
547 if self._base_path == "/":
548 return keys
549 else:
550 # NOTE: PosixStorageProvider does not have the concept of bucket and prefix.
551 # So we drop the base_path from it.
552 return [key.replace(self._base_path, "", 1).lstrip("/") for key in keys]
553
[docs]
554 def is_file(self, path: str) -> bool:
555 path = self._prepend_base_path(path)
556 return os.path.isfile(path)
557
[docs]
558 def rmtree(self, path: str) -> None:
559 path = self._prepend_base_path(path)
560 shutil.rmtree(path)