Coverage for openhcs/io/filemanager.py: 60.9%
220 statements
« prev ^ index » next coverage.py v7.10.3, created at 2025-08-14 05:57 +0000
1"""
2FileManager directory operations.
4This module contains the directory-related methods of the FileManager class,
5including directory listing, existence checking, mkdir, symlink, and mirror operations.
6"""
8import logging
9import os
10from pathlib import Path
11from typing import List, Set, Union, Tuple, Optional, Any
13from openhcs.constants.constants import DEFAULT_IMAGE_EXTENSIONS
14from openhcs.io.base import StorageBackend
15from openhcs.io.exceptions import PathMismatchError, StorageResolutionError
16from openhcs.validation import validate_path_types, validate_backend_parameter
17import traceback
19logger = logging.getLogger(__name__)
21class FileManager:
23 def __init__(self, registry):
24 """
25 Initialize the file manager.
27 Args:
28 registry: Registry for storage backends. Must be provided.
30 Raises:
31 ValueError: If registry is not provided.
33 Note:
34 This class is a backend-agnostic router. It maintains no default backend
35 or fallback behavior, and all state is instance-local and declarative.
36 Each operation must explicitly specify which backend to use.
38 Thread Safety:
39 Each FileManager instance must be scoped to a single execution context.
40 Do NOT share FileManager instances across pipelines or threads.
41 For isolation, create a dedicated registry for each FileManager.
42 """
43 # Validate registry parameter
44 if registry is None: 44 ↛ 45line 44 didn't jump to line 45 because the condition on line 44 was never true
45 raise ValueError("Registry must be provided to FileManager. Default fallback has been removed.")
47 # Store registry
48 self.registry = registry
52 logger.debug("FileManager initialized with registry")
54 def _get_backend(self, backend_name: str) -> StorageBackend:
55 """
56 Get a backend by name.
58 This method uses the instance registry to get the backend instance directly.
59 All FileManagers that use the same registry share the same backend instances.
61 Args:
62 backend_name: Name of the backend to get (e.g., "disk", "memory", "zarr")
64 Returns:
65 The backend instance
67 Raises:
68 StorageResolutionError: If the backend is not found in the registry
70 Thread Safety:
71 Backend instances are shared across all FileManager instances that use
72 the same registry. This ensures shared state (especially for memory backend).
73 """
74 # Normalize backend name
75 backend_name = backend_name.lower()
77 if backend_name is None: 77 ↛ 78line 77 didn't jump to line 78 because the condition on line 77 was never true
78 raise StorageResolutionError(f"Backend '{backend_name}' not found in registry")
80 try:
81 # Get the backend instance from the registry dictionary
82 if backend_name not in self.registry: 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true
83 raise KeyError(f"Backend '{backend_name}' not found in registry")
85 # Return the backend instance directly
86 return self.registry[backend_name]
87 except Exception as e:
88 raise StorageResolutionError(f"Failed to get backend '{backend_name}': {e}") from e
90 def load(self, file_path: Union[str, Path], backend: str, **kwargs) -> Any:
91 """
92 Load data from a file using the specified backend.
94 This method assumes the file path is already backend-compatible and performs no inference or fallback.
95 All semantic validation and file format decoding must occur within the backend.
97 Args:
98 file_path: Path to the file to load (str or Path)
99 backend: Backend enum to use for loading (StorageBackendType.DISK, etc.) — POSITIONAL argument
100 **kwargs: Additional keyword arguments passed to the backend's load method
102 Returns:
103 Any: The loaded data object
105 Raises:
106 StorageResolutionError: If the backend is not supported or load fails
107 """
109 try:
110 backend_instance = self._get_backend(backend)
111 return backend_instance.load(file_path, **kwargs)
112 except StorageResolutionError: # Allow specific backend errors to propagate
113 raise
114 except Exception as e:
115 logger.error(f"Unexpected error during load from {file_path} with backend {backend}: {e}", exc_info=True)
116 raise StorageResolutionError(
117 f"Failed to load file at {file_path} using backend '{backend}'"
118 ) from e
120 def save(self, data: Any, output_path: Union[str, Path], backend: str, **kwargs) -> None:
121 """
122 Save data to a file using the specified backend.
124 This method performs no semantic transformation, format inference, or fallback logic.
125 It assumes the output path and data are valid and structurally aligned with the backend’s expectations.
127 Args:
128 data: The data object to save (e.g., np.ndarray, torch.Tensor, dict, etc.)
129 output_path: Destination path to write to (str or Path)
130 backend: Backend enum to use for saving (StorageBackendType.DISK, etc.) — POSITIONAL argument
131 **kwargs: Additional keyword arguments passed to the backend's save method
133 Raises:
134 StorageResolutionError: If the backend is not supported or save fails
135 """
137 try:
138 backend_instance = self._get_backend(backend)
139 backend_instance.save(data, output_path, **kwargs)
140 except StorageResolutionError: # Allow specific backend errors to propagate if they are StorageResolutionError
141 raise
142 except Exception as e:
143 logger.error(f"Unexpected error during save to {output_path} with backend {backend}: {e}", exc_info=True)
144 raise StorageResolutionError(
145 f"Failed to save data to {output_path} using backend '{backend}'"
146 ) from e
148 def load_batch(self, file_paths: List[Union[str, Path]], backend: str, **kwargs) -> List[Any]:
149 """
150 Load multiple files using the specified backend.
152 Args:
153 file_paths: List of file paths to load
154 backend: Backend to use for loading
155 **kwargs: Additional keyword arguments passed to the backend's load_batch method
157 Returns:
158 List of loaded data objects in the same order as file_paths
160 Raises:
161 StorageResolutionError: If the backend is not supported or load fails
162 """
163 try:
164 backend_instance = self._get_backend(backend)
165 return backend_instance.load_batch(file_paths, **kwargs)
166 except StorageResolutionError:
167 raise
168 except Exception as e:
169 logger.error(f"Unexpected error during batch load with backend {backend}: {e}", exc_info=True)
170 raise StorageResolutionError(
171 f"Failed to load batch of {len(file_paths)} files using backend '{backend}'"
172 ) from e
174 def save_batch(self, data_list: List[Any], output_paths: List[Union[str, Path]], backend: str, **kwargs) -> None:
175 """
176 Save multiple data objects using the specified backend.
178 Args:
179 data_list: List of data objects to save
180 output_paths: List of destination paths (must match length of data_list)
181 backend: Backend to use for saving
182 **kwargs: Additional keyword arguments passed to the backend's save_batch method
184 Raises:
185 StorageResolutionError: If the backend is not supported or save fails
186 ValueError: If data_list and output_paths have different lengths
187 """
188 try:
189 backend_instance = self._get_backend(backend)
190 backend_instance.save_batch(data_list, output_paths, **kwargs)
191 except StorageResolutionError:
192 raise
193 except Exception as e:
194 logger.error(f"Unexpected error during batch save with backend {backend}: {e}", exc_info=True)
195 raise StorageResolutionError(
196 f"Failed to save batch of {len(data_list)} files using backend '{backend}'"
197 ) from e
199 def list_image_files(self, directory: Union[str, Path], backend: str,
200 pattern: str = None, extensions: Set[str] = DEFAULT_IMAGE_EXTENSIONS, recursive: bool = False) -> List[str]:
201 """
202 List all image files in a directory using the specified backend.
204 This method performs no semantic validation, normalization, or naming enforcement on the input path.
205 It assumes the caller has provided a valid, backend-compatible path and merely dispatches it for execution.
207 Note: ONLY backend is a POSITIONAL argument. Other parameters may remain as kwargs.
209 Args:
210 directory: Directory to search (str or Path)
211 backend: Backend to use for listing ('disk', 'memory', 'zarr') - POSITIONAL
212 pattern: Pattern to filter files (e.g., "*.tif") - can be keyword arg
213 extensions: Set of file extensions to filter by - can be keyword arg
214 recursive: Whether to search recursively - can be keyword arg
216 Returns:
217 List of string paths for image files found
219 Raises:
220 StorageResolutionError: If the backend is not supported
221 TypeError: If directory is not a valid path type
222 PathMismatchError: If the path scheme doesn't match the expected scheme for the backend
223 """
224 # Get backend instance
225 backend_instance = self._get_backend(backend)
227 # List image files and apply natural sorting
228 from openhcs.core.utils import natural_sort
229 files = backend_instance.list_files(str(directory), pattern, extensions, recursive)
230 return natural_sort(files)
233 def list_files(self, directory: Union[str, Path], backend: str,
234 pattern: str = None, extensions: Set[str] = None, recursive: bool = False) -> List[str]:
235 """
236 List all files in a directory using the specified backend.
238 This method performs no semantic validation, normalization, or naming enforcement on the input path.
239 It assumes the caller has provided a valid, backend-compatible path and merely dispatches it for execution.
241 Note: ONLY backend is a POSITIONAL argument. Other parameters may remain as kwargs.
243 Args:
244 directory: Directory to search (str or Path)
245 backend: Backend to use for listing ('disk', 'memory', 'zarr') - POSITIONAL
246 pattern: Pattern to filter files (e.g., "*.txt") - can be keyword arg
247 extensions: Set of file extensions to filter by - can be keyword arg
248 recursive: Whether to search recursively - can be keyword arg
250 Returns:
251 List of string paths for files found
253 Raises:
254 StorageResolutionError: If the backend is not supported
255 TypeError: If directory is not a valid path type
256 PathMismatchError: If the path scheme doesn't match the expected scheme for the backend
257 """
258 # Get backend instance
259 backend_instance = self._get_backend(backend)
261 # List files and apply natural sorting
262 from openhcs.core.utils import natural_sort
263 files = backend_instance.list_files(str(directory), pattern, extensions, recursive)
264 return natural_sort(files)
267 def find_file_recursive(self, directory: Union[str, Path], filename: str, backend: str) -> Union[str, None]:
268 """
269 Find a file recursively in a directory using the specified backend.
271 This is a convenience method that uses list_files with recursive=True and filters for the specific filename.
273 Args:
274 directory: Directory to search (str or Path)
275 filename: Name of the file to find
276 backend: Backend to use for listing ('disk', 'memory', 'zarr') - POSITIONAL
278 Returns:
279 String path to the file if found, None otherwise
281 Raises:
282 StorageResolutionError: If the backend is not supported
283 TypeError: If directory is not a valid path type
284 PathMismatchError: If the path scheme doesn't match the expected scheme for the backend
285 """
286 # List all files recursively
287 all_files = self.list_files(directory, backend, recursive=True)
289 # Filter for the specific filename
290 for file_path in all_files:
291 if Path(file_path).name == filename:
292 return file_path
294 # File not found
295 return None
298 def list_dir(self, path: Union[str, Path], backend: str) -> List[str]:
299 if not isinstance(path, (str, Path)): 299 ↛ 300line 299 didn't jump to line 300 because the condition on line 299 was never true
300 raise TypeError(f"Expected str or Path, got {type(path)}")
302 path = str(path)
303 backend_instance = self._get_backend(backend)
305 try:
306 # Get directory listing and apply natural sorting
307 from openhcs.core.utils import natural_sort
308 entries = backend_instance.list_dir(str(path))
309 return natural_sort(entries)
310 except (FileNotFoundError, NotADirectoryError):
311 # Let these bubble up for structural truth-checking
312 raise
313 except Exception as e:
314 # Optional trace wrapper, no type mutation
315 raise RuntimeError(f"Unexpected failure in list_dir({path}) for backend {backend}") from e
317 def ensure_directory(self, directory: Union[str, Path], backend: str) -> str:
318 """
319 Ensure a directory exists, creating it if necessary.
321 This method performs no semantic validation, normalization, or naming enforcement on the input path.
322 It assumes the caller has provided a valid, backend-compatible path and merely dispatches it for execution.
324 Note: ONLY backend is a POSITIONAL argument. All parameters are required.
326 Args:
327 directory: Directory to ensure exists (str or Path)
328 backend: Backend to use for directory operations ('disk', 'memory', 'zarr') - POSITIONAL
330 Returns:
331 String path to the directory
333 Raises:
334 StorageResolutionError: If the backend is not supported
335 TypeError: If directory is not a valid path type
336 PathMismatchError: If the path scheme doesn't match the expected scheme for the backend
337 """
338 # Get backend instance
339 backend_instance = self._get_backend(backend)
341 # Ensure directory
342 return backend_instance.ensure_directory(str(directory))
346 def exists(self, path: Union[str, Path], backend: str) -> bool:
347 """
348 Check if a path exists.
350 This method performs no semantic validation, normalization, or naming enforcement on the input path.
351 It assumes the caller has provided a valid, backend-compatible path and merely dispatches it for execution.
353 Note: ONLY backend is a POSITIONAL argument. All parameters are required.
355 Args:
356 path: Path to check (str or Path)
357 backend: Backend to use for checking ('disk', 'memory', 'zarr') - POSITIONAL
359 Returns:
360 True if the path exists, False otherwise
362 Raises:
363 StorageResolutionError: If the backend is not supported
364 TypeError: If path is not a valid path type
365 PathMismatchError: If the path scheme doesn't match the expected scheme for the backend
366 """
367 # Get backend instance
368 backend_instance = self._get_backend(backend)
370 # Check if path exists
371 return backend_instance.exists(str(path))
374 def mirror_directory_with_symlinks(
375 self,
376 source_dir: Union[str, Path],
377 target_dir: Union[str, Path],
378 backend: str,
379 recursive: bool = True,
380 overwrite_symlinks_only: bool = False
381 ) -> int:
382 """
383 Mirror a directory structure from source to target and create symlinks to all files.
385 This method performs no semantic validation, normalization, or naming enforcement on the input paths.
386 It assumes the caller has provided valid, backend-compatible paths and merely dispatches them for execution.
388 By default, this method will NOT overwrite existing files. Use overwrite_symlinks_only=True to allow
389 overwriting existing symlinks (but not regular files).
391 Note: ONLY backend is a POSITIONAL argument. Other parameters may remain as kwargs.
393 Args:
394 source_dir: Path to the source directory to mirror (str or Path)
395 target_dir: Path to the target directory where the mirrored structure will be created (str or Path)
396 backend: Backend to use for mirroring ('disk', 'memory', 'zarr') - POSITIONAL
397 recursive: Whether to recursively mirror subdirectories - can be keyword arg
398 overwrite_symlinks_only: If True, allows overwriting existing symlinks but blocks overwriting regular files.
399 If False (default), no overwriting is allowed. - can be keyword arg
401 Returns:
402 int: Number of symlinks created
404 Raises:
405 StorageResolutionError: If the backend is not supported
406 FileExistsError: If target files exist and overwrite_symlinks_only=False, or if trying to overwrite regular files
407 TypeError: If source_dir or target_dir is not a valid path type
408 PathMismatchError: If the path scheme doesn't match the expected scheme for the backend
409 """
410 # Get backend instance
411 backend_instance = self._get_backend(backend)
412 # Mirror the directory structure and create symlinks for files recursively
413 self.ensure_directory(target_dir, backend)
414 try:
415 # Ensure target directory exists
417 # Count symlinks
418 symlink_count = 0
420 # Get all directories under source_dir (including source_dir itself)
422 _, all_files = self.collect_dirs_and_files(source_dir, backend, recursive=True)
424 # 1. Ensure base target exists
425 self.ensure_directory(target_dir, backend)
427 # 2. Symlink all file paths
428 for file_path in all_files:
429 rel_path = Path(file_path).relative_to(Path(source_dir))
430 symlink_path = Path(target_dir) / rel_path
431 self.create_symlink(file_path, str(symlink_path), backend, overwrite_symlinks_only=overwrite_symlinks_only)
432 symlink_count += 1
434 return symlink_count
436 except Exception as e:
437 raise StorageResolutionError(f"Failed to mirror directory {source_dir} to {target_dir} with backend {backend}") from e
439 def create_symlink(
440 self,
441 source_path: Union[str, Path],
442 symlink_path: Union[str, Path],
443 backend: str,
444 overwrite_symlinks_only: bool = False
445 ) -> bool:
446 """
447 Create a symbolic link from source_path to symlink_path.
449 This method performs no semantic validation, normalization, or naming enforcement on the input paths.
450 It assumes the caller has provided valid, backend-compatible paths and merely dispatches them for execution.
452 Note: ONLY backend is a POSITIONAL argument. All parameters are required.
454 Args:
455 source_path: Path to the source file or directory (str or Path)
456 symlink_path: Path where the symlink should be created (str or Path)
457 backend: Backend to use for symlink creation ('disk', 'memory', 'zarr') - POSITIONAL
458 overwrite_symlinks_only: If True, only allow overwriting existing symlinks (not regular files)
460 Returns:
461 bool: True if successful, False otherwise
463 Raises:
464 StorageResolutionError: If the backend is not supported
465 FileExistsError: If target exists and is not a symlink when overwrite_symlinks_only=True
466 VFSTypeError: If source_path or symlink_path cannot be converted to internal path format
467 PathMismatchError: If the path scheme doesn't match the expected scheme for the backend
468 """
469 # Get backend instance
470 backend_instance = self._get_backend(backend)
472 # Check if target exists and handle overwrite policy
473 try:
474 if backend_instance.exists(str(symlink_path)): 474 ↛ 475line 474 didn't jump to line 475 because the condition on line 474 was never true
475 if overwrite_symlinks_only:
476 # Check if existing target is a symlink
477 if not self.is_symlink(symlink_path, backend):
478 raise FileExistsError(
479 f"Target exists and is not a symlink (overwrite_symlinks_only=True): {symlink_path}"
480 )
481 # Target is a symlink, allow overwrite
482 backend_instance.create_symlink(str(source_path), str(symlink_path), overwrite=True)
483 else:
484 # No overwrite allowed
485 raise FileExistsError(f"Target already exists: {symlink_path}")
486 else:
487 # Target doesn't exist, create new symlink
488 backend_instance.create_symlink(str(source_path), str(symlink_path), overwrite=False)
490 return True
491 except FileExistsError:
492 # Re-raise FileExistsError from our check or from backend
493 raise
494 except Exception as e:
495 raise StorageResolutionError(
496 f"Failed to create symlink from {source_path} to {symlink_path} with backend {backend}"
497 ) from e
499 def delete(self, path: Union[str, Path], backend: str, recursive: bool = False) -> bool:
500 """
501 Delete a file or directory.
503 This method performs no semantic validation, normalization, or naming enforcement on the input path.
504 It assumes the caller has provided a valid, backend-compatible path and merely dispatches it for execution.
506 Note: ONLY backend is a POSITIONAL argument. All parameters are required.
508 Args:
509 path: Path to the file or directory to delete (str or Path)
510 backend: Backend to use for deletion ('disk', 'memory', 'zarr') - POSITIONAL
512 Returns:
513 True if successful, False otherwise
515 Raises:
516 StorageResolutionError: If the backend is not supported
517 FileNotFoundError: If the file does not exist
518 TypeError: If the path is not a valid path type
519 """
520 # Get backend instance
521 backend_instance = self._get_backend(backend)
523 # Delete the file or directory
524 try:
525 # No virtual path conversion needed
526 return backend_instance.delete(str(path))
527 except Exception as e:
528 raise StorageResolutionError(
529 f"Failed to delete {path} with backend {backend}"
530 ) from e
532 def delete_all(self, path: Union[str, Path], backend: str) -> bool:
533 """
534 Recursively delete a file, symlink, or directory at the given path.
536 This method performs no fallback, coercion, or resolution — it dispatches to the backend.
537 All resolution and deletion behavior must be encoded in the backend's `delete_all()` method.
539 Args:
540 path: The path to delete
541 backend: The backend key (e.g., 'disk', 'memory', 'zarr')
543 Returns:
544 True if successful
546 Raises:
547 StorageResolutionError: If the backend operation fails
548 FileNotFoundError: If the path does not exist
549 TypeError: If the path is not a str or Path
550 """
551 backend_instance = self._get_backend(backend)
552 path_str = str(path)
554 try:
555 backend_instance.delete_all(path_str)
556 return True
557 except Exception as e:
558 raise StorageResolutionError(
559 f"Failed to delete_all({path_str}) using backend '{backend}'"
560 ) from e
563 def copy(self, source_path: Union[str, Path], dest_path: Union[str, Path], backend: str) -> bool:
564 """
565 Copy a file, directory, or symlink from source_path to dest_path using the given backend.
567 - Will NOT overwrite existing files/directories.
568 - Handles symlinks as first-class objects (not dereferenced).
569 - Raises on broken links or mismatched structure.
571 Raises:
572 FileExistsError: If destination exists
573 FileNotFoundError: If source does not exist
574 StorageResolutionError: On backend failure
575 """
576 backend_instance = self._get_backend(backend)
578 try:
579 # Prevent overwriting
580 if backend_instance.exists(dest_path):
581 raise FileExistsError(f"Destination already exists: {dest_path}")
583 # Ensure destination parent exists
584 dest_parent = Path(dest_path).parent
585 self.ensure_directory(dest_parent, backend)
587 # Delegate to backend-native copy
588 return backend_instance.copy(str(source_path), str(dest_path))
589 except Exception as e:
590 raise StorageResolutionError(
591 f"Failed to copy from {source_path} to {dest_path} on backend {backend}"
592 ) from e
595 def move(self, source_path: Union[str, Path], dest_path: Union[str, Path], backend: str,
596 replace_symlinks: bool = False) -> bool:
597 """
598 Move a file, directory, or symlink from source_path to dest_path.
600 - Will NOT overwrite by default.
601 - Preserves symbolic identity (moves links as links).
602 - Uses backend-native move if available.
603 - Can optionally replace existing symlinks when replace_symlinks=True.
605 Args:
606 source_path: Source file or directory path
607 dest_path: Destination file or directory path
608 backend: Backend to use for the operation
609 replace_symlinks: If True, allows overwriting existing symlinks at destination.
610 If False (default), raises FileExistsError if destination exists.
612 Raises:
613 FileExistsError: If destination exists and replace_symlinks=False, or if
614 destination exists and is not a symlink when replace_symlinks=True
615 FileNotFoundError: If source is missing
616 StorageResolutionError: On backend failure
617 """
618 backend_instance = self._get_backend(backend)
620 try:
621 # Handle destination existence based on replace_symlinks setting
622 if backend_instance.exists(dest_path): 622 ↛ 623line 622 didn't jump to line 623 because the condition on line 622 was never true
623 if replace_symlinks:
624 # Check if destination is a symlink
625 if backend_instance.is_symlink(dest_path):
626 logger.debug("Destination is a symlink, removing before move: %s", dest_path)
627 backend_instance.delete(dest_path)
628 else:
629 # Destination exists but is not a symlink
630 raise FileExistsError(f"Destination already exists and is not a symlink: {dest_path}")
631 else:
632 # replace_symlinks=False, don't allow any overwriting
633 raise FileExistsError(f"Destination already exists: {dest_path}")
635 dest_parent = Path(dest_path).parent
636 self.ensure_directory(dest_parent, backend)
637 return backend_instance.move(str(source_path), str(dest_path))
639 except Exception as e:
640 raise StorageResolutionError(
641 f"Failed to move from {source_path} to {dest_path} on backend {backend}"
642 ) from e
644 def collect_dirs_and_files(
645 self,
646 base_dir: Union[str, Path],
647 backend: str,
648 recursive: bool = True
649 ) -> Tuple[List[str], List[str]]:
650 """
651 Collect all valid directories and files starting from base_dir using breadth-first traversal.
653 Returns:
654 (dirs, files): Lists of string paths for directories and files
655 """
656 from collections import deque
658 base_dir = str(base_dir)
659 # Use deque for breadth-first traversal (FIFO instead of LIFO)
660 queue = deque([base_dir])
661 dirs: List[str] = []
662 files: List[str] = []
664 while queue:
665 current_path = queue.popleft() # FIFO for breadth-first
667 try:
668 entries = self.list_dir(current_path, backend)
669 dirs.append(current_path)
670 except (NotADirectoryError, FileNotFoundError):
671 files.append(current_path)
672 continue
673 except Exception as e:
674 print(f"[collect_dirs_and_files] Unexpected error at {current_path}: {type(e).__name__} — {e}")
675 continue # Fail-safe: skip unexpected issues
677 if entries is None: 677 ↛ 679line 677 didn't jump to line 679 because the condition on line 677 was never true
678 # Defensive fallback — entries must be iterable
679 print(f"[collect_dirs_and_files] WARNING: list_dir() returned None at {current_path}")
680 continue
682 for entry in entries:
683 full_path = str(Path(current_path) / entry)
684 try:
685 self.list_dir(full_path, backend)
686 dirs.append(full_path)
687 if recursive: 687 ↛ 682line 687 didn't jump to line 682 because the condition on line 687 was always true
688 queue.append(full_path) # Add to end of queue for breadth-first
689 except (NotADirectoryError, FileNotFoundError):
690 files.append(full_path)
691 except Exception as e:
692 print(f"[collect_dirs_and_files] Skipping {full_path}: {type(e).__name__} — {e}")
693 continue
695 # Apply natural sorting to both dirs and files before returning
696 from openhcs.core.utils import natural_sort
697 return natural_sort(dirs), natural_sort(files)
699 def is_file(self, path: Union[str, Path], backend: str) -> bool:
700 """
701 Check if a given path is a file using the specified backend.
703 Args:
704 path: Path to check (raw string or Path)
705 backend: Backend key ('disk', 'memory', 'zarr') — must be positional
707 Returns:
708 bool: True if the path is a file, False otherwise (including if path doesn't exist)
709 """
710 try:
711 backend_instance = self._get_backend(backend)
712 return backend_instance.is_file(path)
713 except Exception:
714 # Return False for any error (file not found, is a directory, backend issues)
715 return False
717 def is_dir(self, path: Union[str, Path], backend: str) -> bool:
718 """
719 Check if a given path is a directory using the specified backend.
721 Args:
722 path: Path to check (raw string or Path)
723 backend: Backend key ('disk', 'memory', 'zarr') — must be positional
725 Returns:
726 bool: True if the path is a directory, False if it's a file or doesn't exist
728 Raises:
729 StorageResolutionError: If resolution fails or backend misbehaves
730 """
731 try:
732 backend_instance = self._get_backend(backend)
733 return backend_instance.is_dir(path)
734 except (FileNotFoundError, NotADirectoryError):
735 # Return False for files or non-existent paths instead of raising
736 return False
737 except Exception as e:
738 raise StorageResolutionError(
739 f"Failed to check if {path} is a directory with backend '{backend}'"
740 ) from e
742 def is_symlink(self, path: Union[str, Path], backend: str) -> bool:
743 """
744 Check if a given path is a symbolic link using the specified backend.
746 Args:
747 path: Path to check (raw string or Path)
748 backend: Backend key ('disk', 'memory', 'zarr') — must be positional
750 Returns:
751 bool: True if the path is a symbolic link, False otherwise (including if path doesn't exist)
752 """
753 try:
754 backend_instance = self._get_backend(backend)
755 return backend_instance.is_symlink(str(path))
756 except Exception:
757 # Return False for any error (file not found, not a symlink, backend issues)
758 return False