"""
Workspace context after successful authentication.
"""
import logging
import threading
import warnings
from typing import Dict, List, Optional
from .exceptions import FilespaceAlreadyLinkedError
from .filespace import Filespace
from .filespace_models import FilespaceInfo, SyncMode
logger = logging.getLogger(__name__)
# Wire discriminator of filespace-internal-error notifications on the native notification
# channel — must match ToString at the C ABI gate in Sdk/Api/lucid_api_daemon.cpp.
_NOTIFICATION_FILESPACE_INTERNAL_ERROR = "filespace_internal_error"
[docs]
class Workspace:
"""
Workspace context after successful authentication.
Provides access to filespace operations within the authenticated workspace.
Returned by ``Daemon.authenticate()``.
Multiple filespaces can be linked concurrently — each
``link_filespace()`` call returns an independent ``Filespace`` that
stays usable until it is unlinked.
Example:
.. code-block:: python
credentials = ServiceAccountCredentials(token)
client.login(credentials)
workspace = client.get_workspace(client.list_workspaces()[0].id)
print(workspace.id, workspace.name)
"""
def __init__(self, native_daemon, workspace_id: str, workspace_name: str, notification_watcher=None):
"""
Initialize workspace context.
Args:
native_daemon: Native runtime wrapper (internal use)
workspace_id: Workspace ID
workspace_name: Workspace name
notification_watcher: The client's daemon-notification watcher;
this workspace subscribes itself as its handler (internal use)
Note: This constructor is called internally by ``Client.login()``.
Users should not construct ``Workspace`` objects directly.
:meta private-args: native_daemon
"""
self._native_daemon = native_daemon
self._id = workspace_id
self._name = workspace_name
self._linked_filespaces: Dict[str, Filespace] = {} # filespace id -> Filespace
self._aliases: Dict[str, str] = {} # name-or-id used at link time -> filespace id
# Single lock for every read or write of `_linked_filespaces` /
# `_aliases` — link/unlink/stop/iteration. The GIL atomises single
# bytecodes but not multi-step patterns (the comprehension reassignment
# in `_on_filespace_unlinked`, the find-then-link check in
# `link_filespace`, dict iteration while another thread mutates), so
# those need explicit serialisation. RLock because `stop()` holds the
# lock while calling `filespace.unlink()`, which re-enters
# `_on_filespace_unlinked` on the same thread.
self._link_lock = threading.RLock()
if notification_watcher is not None:
notification_watcher.set_handler(self._on_daemon_notification)
@property
def id(self) -> str:
"""Get the workspace ID."""
return self._id
@property
def name(self) -> str:
"""Get the workspace name."""
return self._name
@property
def linked_filespaces(self) -> List[Filespace]:
"""Get the currently linked ``Filespace`` objects."""
with self._link_lock:
return [fs for fs in self._linked_filespaces.values() if fs.is_linked]
[docs]
def list_filespaces(self) -> List[FilespaceInfo]:
"""
List all filespaces in this workspace.
Returns:
List of FilespaceInfo objects with id, name, and created timestamp.
Raises:
ConnectionError: If LucidLink services are unreachable
AuthenticationError: If access token expired
RuntimeError: If the client is not running
Example:
.. code-block:: python
filespaces = workspace.list_filespaces()
for fs in filespaces:
print(f"{fs.name}")
"""
return [
FilespaceInfo(id=d["id"], name=d["name"], created=d["created"])
for d in self._native_daemon.list_filespaces(self.id)
]
[docs]
def link_filespace(
self,
name: Optional[str] = None,
id: Optional[str] = None,
root_path: str = "/",
sync_mode: SyncMode = SyncMode.SYNC_ALL,
):
"""
Link to a filespace in this workspace.
You must provide either ``name`` OR ``id``, but not both.
Multiple filespaces can be linked at the same time: linking a second
filespace does NOT unlink the first one. Each linked filespace runs a
full client stack — expect roughly one disk cache
default 1024 MB) plus a set of worker threads per link.
Linking a filespace that is already linked is idempotent and returns
the existing live ``Filespace`` object. One nuance: in the brief
window between a link's internal-error teardown and the watcher thread
dispatching its notification (see :class:`FilespaceInternalError`) the
stale entry is still returned; once the notification marks it, it is
dropped and a retry links fresh.
Args:
name: Filespace name.
.. deprecated::
Pass ``id`` instead — a filespace name is mutable while
its id is stable, so a script that worked yesterday
silently links to nothing (or the wrong filespace) after
a rename.
id: Filespace ID (recommended). Stable for the lifetime of the
filespace.
root_path: Mount point path (default: ``"/"``)
sync_mode: Controls automatic sync on close. ``SYNC_ALL`` (default)
calls ``sync_all()`` before unlinking. ``SYNC_NONE`` skips
automatic sync — caller must call ``sync_all()`` explicitly.
Returns:
``Filespace`` object for filesystem operations
Raises:
ValueError: If neither ``name`` nor ``id`` provided, or both provided
FileNotFoundError: If filespace not found
PermissionDeniedError: If service account lacks access
RuntimeError: If the client is not running or not authenticated
Example:
.. code-block:: python
# Link by ID (recommended — stable across renames)
fs = workspace.link_filespace(id="fs-uuid-12345")
# Link a second filespace — both stay usable concurrently
fs2 = workspace.link_filespace(id="fs-uuid-67890")
# Using as context manager (auto sync + unlink on exit)
with workspace.link_filespace(id="fs-uuid-12345") as fs:
fs.fs.write_file("/file.txt", b"data")
# sync_all() + unlink() called automatically
# Disable auto-sync
fs = workspace.link_filespace(
id="fs-uuid-12345", sync_mode=SyncMode.SYNC_NONE)
# Deprecated: link by name (emits DeprecationWarning)
fs = workspace.link_filespace(name="production-data")
"""
if name is None and id is None:
raise ValueError("Must provide either 'name' or 'id'")
if name is not None and id is not None:
raise ValueError("Cannot provide both 'name' and 'id'")
if name is not None:
warnings.warn(
"The 'name' argument of link_filespace() is deprecated; pass 'id' "
"instead — a filespace name is mutable while its id is stable.",
DeprecationWarning,
stacklevel=2,
)
identifier = name if name is not None else id
# The lock spans both the cache check and the native call so two threads
# racing on the same identifier serialize: the second sees the entry the
# first published and returns it. Concurrent links on different
# identifiers serialize within Python too — the native LinkFilespace path
# already serializes through lifecycleMutex, so this isn't extra cost.
with self._link_lock:
existing = self._find_linked(identifier)
if existing is not None:
return existing
try:
native_fs = self._native_daemon.link_filespace(
workspace_id=self.id,
filespace_name=name if name else "",
filespace_id=id if id else "",
root_path=root_path,
)
except FilespaceAlreadyLinkedError as e:
# Alias miss: the filespace is already linked but under an
# identifier we haven't seen (e.g. linked by id, requested by
# name). The daemon classified this by exception type, so we
# know it's a double link — not a network or auth failure —
# and can resolve it to the existing live Filespace.
recovered = self._recover_alias_miss(identifier, e)
if recovered is not None:
return recovered
raise
filespace = Filespace(
native_linked_fs=native_fs,
native_daemon=self._native_daemon,
workspace_id=self._id,
workspace_name=self._name,
id=native_fs.id,
full_name=native_fs.name,
sync_mode=sync_mode,
on_unlink=self._on_filespace_unlinked,
)
self._linked_filespaces[filespace.id] = filespace
self._aliases[identifier] = filespace.id
return filespace
[docs]
def unlink_filespace(self, linked) -> None:
"""Unlink the link identified by a native ``LinkedFilespace`` handle.
Forwarded from ``Daemon.unlink_filespace``. When the handle maps to a
tracked ``Filespace`` the teardown goes through it, so the link
bookkeeping (the ``_linked_filespaces`` / ``_aliases`` entries) is
dropped via the usual unlink callback. An untracked handle falls back
to a direct native unlink so a live link is never leaked.
Users should call ``filespace.unlink()`` instead.
"""
with self._link_lock:
filespace = self._linked_filespaces.get(linked.id)
if filespace is not None:
filespace.unlink()
else:
self._native_daemon.unlink_filespace(linked)
def _on_daemon_notification(self, notification: Dict) -> None:
"""Apply one daemon push notification (arrives on the client's watcher
thread): a filespace internal error unlinks the filespace locally."""
if notification.get("type") != _NOTIFICATION_FILESPACE_INTERNAL_ERROR:
return
filespace_id = notification.get("filespace_id", "")
logger.warning("Filespace %s: %s", filespace_id, notification.get("reason", ""))
filespace = self._get_linked_by_id(filespace_id)
if filespace is not None:
filespace._mark_unlinked()
def _get_linked_by_id(self, filespace_id: str) -> Optional[Filespace]:
"""Thread-safe registry lookup by canonical filespace id — no alias
logic and no liveness filter (the notification handler wants the entry
even though it is about to flip ``is_linked``)."""
with self._link_lock:
return self._linked_filespaces.get(filespace_id)
def _find_linked(self, identifier: str) -> Optional[Filespace]:
"""Resolve an identifier (link-time name or filespace id) to a live link.
Caller must hold ``_link_lock``.
"""
fs_id = self._aliases.get(identifier)
if fs_id is None and identifier in self._linked_filespaces:
fs_id = identifier
if fs_id is None:
return None
filespace = self._linked_filespaces.get(fs_id)
if filespace is not None and filespace.is_linked:
return filespace
return None
def _recover_alias_miss(
self, identifier: str, error: FilespaceAlreadyLinkedError
) -> Optional[Filespace]:
"""Map an already-linked error back to the live Filespace and register
the new alias. Returns None (caller re-raises) if the daemon's id is
absent or no longer tracked — a state we can't safely recover from.
Caller must hold ``_link_lock``.
"""
fs_id = error.filespace_id
if not fs_id:
return None
filespace = self._linked_filespaces.get(fs_id)
if filespace is None or not filespace.is_linked:
return None
self._aliases[identifier] = fs_id
return filespace
def _on_filespace_unlinked(self, filespace_id: str) -> None:
"""Callback from Filespace.unlink — drop the entry and any aliases that
point at this id so a long-running service doesn't accumulate zombies."""
with self._link_lock:
self._linked_filespaces.pop(filespace_id, None)
self._aliases = {k: v for k, v in self._aliases.items() if v != filespace_id}
[docs]
def stop(self) -> None:
"""Stop workspace — unlinks all linked filespaces.
If a filespace's ``sync_mode`` is ``SYNC_ALL``, ``sync_all()`` is called
before it is unlinked. Per-filespace unlink errors are swallowed so one
failing link never blocks the others. Safe to call multiple times.
"""
# `filespace.unlink()` re-enters `_on_filespace_unlinked` on the same
# thread (the lock is an RLock so the re-entry is OK). Holding the lock
# across the iteration also stops a concurrent `link_filespace` from
# publishing a new entry that would survive the clear below.
with self._link_lock:
for filespace in list(self._linked_filespaces.values()):
try:
filespace.unlink()
except Exception:
pass
self._linked_filespaces.clear()
self._aliases.clear()
def __repr__(self) -> str:
return f"Workspace(id='{self.id}', name='{self.name}')"