Source code for lucidlink.storage

"""
Storage configuration for LucidLink Python daemon.

This module provides two operational modes for managing daemon files:
- ``SANDBOXED``: Files in temporary directory, always cleaned up on exit
- ``PHYSICAL``: Files in ``.lucid`` subfolder, optional persistence

IMPORTANT: Directory Structure
===============================
The native C++ daemon automatically manages the directory structure:
1. Python passes the BASE directory to the daemon (e.g., C:/work/script/)
2. Daemon automatically adds the .lucid subdirectory
3. Daemon creates per-filespace UUID subdirectories for isolation

Example with PHYSICAL mode running from ``C:/work/script/``::

    1. StorageConfig.get_root_path() returns: C:/work/script/
    2. Python passes to daemon: "C:/work/script" (forward slashes)
    3. Native daemon creates directory structure:
       - C:/work/script/.lucid/{filespace1-uuid}/node.cfg
       - C:/work/script/.lucid/{filespace1-uuid}/metadb/
       - C:/work/script/.lucid/{filespace1-uuid}/cache/
    4. If you link to a second filespace in the same script:
       - C:/work/script/.lucid/{filespace2-uuid}/node.cfg
       - C:/work/script/.lucid/{filespace2-uuid}/metadb/
       - C:/work/script/.lucid/{filespace2-uuid}/cache/

This prevents file clashes when multiple daemon instances link to different
filespaces from the same script.

**Note**: Python passes paths to the daemon using forward slashes (generic
format) for cross-platform compatibility, matching the behavior of Android/iOS
which use boost::filesystem::generic_string().
"""

from enum import Enum
from pathlib import Path
import tempfile
import atexit
import shutil
from typing import Optional


[docs] class StorageMode(Enum): """Storage mode for daemon operational files.""" PHYSICAL = "physical" """Files in ``.lucid`` subfolder of script directory.""" SANDBOXED = "sandboxed" """Files in temp directory, always cleaned up."""
[docs] class StorageConfig: """ Configuration for daemon storage mode and file locations. Args: mode: Storage mode (``PHYSICAL`` or ``SANDBOXED``) persist_on_exit: If ``False``, clean up files when daemon stops. Only applies to ``PHYSICAL`` mode; ``SANDBOXED`` always cleans up. root_path: Override root path for files (only for ``PHYSICAL`` mode). If ``None``, uses current working directory. Example: .. code-block:: python # Sandboxed mode (default) - temp directory, always cleaned up config = StorageConfig() # Physical mode with cleanup config = StorageConfig(mode=StorageMode.PHYSICAL) # Physical mode with persistence config = StorageConfig( mode=StorageMode.PHYSICAL, persist_on_exit=True ) # Custom root path config = StorageConfig( mode=StorageMode.PHYSICAL, root_path=Path("D:/lucid_data") ) """ def __init__( self, mode: StorageMode = StorageMode.SANDBOXED, persist_on_exit: bool = False, root_path: Optional[Path] = None, ): self.mode = mode self.persist_on_exit = persist_on_exit self._root_path = root_path self._temp_dir: Optional[Path] = None self._cleanup_registered = False
[docs] def get_root_path(self) -> Path: """ Get the root path for daemon files. Returns: Path to .lucid directory where daemon will create per-filespace UUID subdirectories for isolation. """ if self.mode == StorageMode.PHYSICAL: if self._root_path: return self._root_path / ".lucid" # Use script's execution directory return Path.cwd() / ".lucid" else: # SANDBOXED if self._temp_dir is None: self._temp_dir = Path(tempfile.mkdtemp(prefix="lucid_")) if not self._cleanup_registered: atexit.register(self._cleanup) self._cleanup_registered = True # For sandboxed mode, use .lucid subdirectory return self._temp_dir / ".lucid"
def _cleanup(self): """Clean up sandbox directory on exit.""" if self._temp_dir and self._temp_dir.exists(): try: shutil.rmtree(self._temp_dir, ignore_errors=True) except Exception: # Ignore cleanup errors during exit pass
[docs] def should_cleanup(self) -> bool: """ Check if files should be cleaned up on daemon stop. Returns: True if files should be cleaned up """ if self.mode == StorageMode.SANDBOXED: return True # Always cleanup in sandboxed mode else: return not self.persist_on_exit