"""Filespace context after successful link.
Provides access to filesystem operations and external file management
on a linked LucidLink filespace.
"""
from typing import Optional
from .connect import ConnectManager
from .filesystem import Filesystem
from .filespace_models import SyncMode
[docs]
class Filespace:
"""
Filespace context after successful link.
Provides identity, filesystem access via the ``fs`` property, and
external file operations via the ``connect`` property.
Example:
.. code-block:: python
# Context manager for automatic sync and unlink on exit
with workspace.link_filespace(name="data") as fs:
fs.fs.write_file("/file.txt", b"data")
# sync_all() + unlink() called automatically
# Manual lifecycle
filespace = workspace.link_filespace(name="production-data")
entries = filespace.fs.read_dir("/")
filespace.fs.create_dir("/new-folder")
with filespace.fs.open("/file.txt", "wb") as f:
f.write(b"data")
"""
def __init__(
self,
native_fs,
native_daemon,
workspace_id: str,
workspace_name: str,
id: str,
full_name: str,
sync_mode: SyncMode = SyncMode.SYNC_ALL,
):
"""
Initialize filespace context.
Args:
native_fs: Native filesystem wrapper (internal use)
native_daemon: Native daemon instance (internal use)
workspace_id: Workspace identifier
workspace_name: Workspace name
id: Filespace unique identifier
full_name: Full filespace name (may include workspace prefix)
sync_mode: Controls automatic sync on close (default: ``SYNC_ALL``)
Note: This constructor is called internally by ``Workspace.link_filespace()``.
Users should not construct ``Filespace`` objects directly.
"""
self._native = native_fs
self._native_daemon = native_daemon
self._workspace_id = workspace_id
self._workspace_name = workspace_name
self._id = id
self._full_name = full_name
self._sync_mode = sync_mode
self._linked = True
self._connect: Optional[ConnectManager] = None
self._fs = Filesystem(native_fs)
@property
def workspace_id(self) -> str:
"""Get the workspace ID."""
return self._workspace_id
@property
def workspace_name(self) -> str:
"""Get the workspace name."""
return self._workspace_name
@property
def id(self) -> str:
"""Get the filespace ID."""
return self._id
@property
def name(self) -> str:
"""Get the short filespace name (first segment before dot)."""
return self._full_name.split(".", 1)[0]
@property
def full_name(self) -> str:
"""Get the full filespace name."""
return self._full_name
@property
def fs(self) -> Filesystem:
"""Get the ``Filesystem`` interface for file and directory operations.
Returns:
``Filesystem`` object providing ``read_dir``, ``write_file``, ``open``,
and all other filesystem operations.
Example:
.. code-block:: python
filespace.fs.read_dir("/")
filespace.fs.write_file("/hello.txt", b"world")
"""
return self._fs
@property
def connect(self) -> ConnectManager:
"""Get the ``ConnectManager`` interface for external file operations.
Provides access to data store management and external file linking.
External files are read-only S3 objects mapped 1:1 as files in the filespace.
Returns:
``ConnectManager`` for data store and external file operations
Raises:
RuntimeError: If Connect is not available (e.g., filespace version too old)
Example:
.. code-block:: python
connect = filespace.connect
connect.add_data_store("my-store", S3DataStoreConfig(...))
connect.link_file("/data/file.csv", "my-store", "path/to/file.csv")
"""
if self._connect is None:
native_connect = self._native_daemon.get_connect()
self._connect = ConnectManager(native_connect)
return self._connect
[docs]
def sync_all(self) -> None:
"""
Synchronize all pending changes to the hub.
Flushes all pending metadata and data changes to ensure they are
propagated to LucidHub and become visible to other clients.
Covers all subsystems — filesystem operations, Connect/external
file changes, and any other pending metadata updates.
Call this method after write operations (create, modify, delete)
to guarantee changes are committed before reading them back or
expecting them to be visible to other clients.
Example:
.. code-block:: python
filespace.fs.write_file("/test.txt", b"data")
filespace.sync_all() # Ensure write is committed
# Now other clients can see the file
Raises:
RuntimeError: If sync fails
"""
self._native.sync_all()
[docs]
def unlink(self) -> None:
"""
Unlink from this filespace.
If ``sync_mode`` is ``SYNC_ALL`` (default), automatically calls ``sync_all()``
before unlinking to ensure pending changes are propagated to the hub.
After calling this method, the filespace object becomes invalid and
cannot be used for filesystem operations.
Safe to call multiple times — subsequent calls are no-ops.
Raises:
RuntimeError: If unlink fails
"""
if not self._linked:
return
if self._sync_mode == SyncMode.SYNC_ALL:
try:
self.sync_all()
except Exception:
pass
self._native_daemon.unlink_filespace()
self._linked = False
def __enter__(self) -> "Filespace":
"""Enter context manager — returns self."""
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
"""Exit context manager — unlinks from filespace.
Calls ``sync_all()`` first if ``sync_mode`` is ``SYNC_ALL``.
"""
self.unlink()
def __repr__(self) -> str:
return f"Filespace(workspace='{self._workspace_name}', id='{self.id}', name='{self.name}')"