"""LucidLink Python Library - Client.
Top-level entry point for the LucidLink SDK. Each ``Client`` is fully
independent, so multiple clients can coexist in one process and operate on
different workspaces concurrently.
"""
import logging
import shutil
import threading
import weakref
from dataclasses import dataclass
from typing import Dict, List, Optional
from . import lucidlink_native
from .credentials import ServiceAccountCredentials
from .exceptions import AuthenticationError, ClientError
from .filespace_models import CacheConfig
from .storage import StorageConfig, StorageMode
from .workspace import Workspace
from .workspace_models import WorkspaceInfo
logger = logging.getLogger(__name__)
class _NotificationWatcher:
"""Owns the native notification channel and the thread parked in it.
The subscriber is held weakly: the thread must not keep the ``Workspace``
(and through it the native daemon) alive, because the native daemon's own
stop/destruction is what ends the channel — and with it this thread. The
thread owns the channel exclusively and releases it on the way out; nobody
else may touch it (see ``NotificationChannel.destroy``).
"""
def __init__(self, native_daemon) -> None:
self._channel = native_daemon.open_notification_channel()
self._handler: Optional[weakref.WeakMethod] = None
self._thread = threading.Thread(
target=self._run,
name="lucidlink-notification-watcher",
daemon=True,
)
self._thread.start()
def set_handler(self, bound_method) -> None:
"""Subscribe a bound method; replaces any previous subscriber."""
self._handler = weakref.WeakMethod(bound_method)
def is_alive(self) -> bool:
return self._thread.is_alive()
def join(self, timeout: float) -> None:
"""Join the watcher thread; a no-op from the watcher thread itself."""
if self._thread is threading.current_thread():
return
self._thread.join(timeout)
def _run(self) -> None:
try:
while True:
notification = self._channel.next() # parks off-GIL
if notification is None:
return # channel ended: the daemon stopped or was destroyed
handler = self._handler() if self._handler is not None else None
if handler is None:
continue # no subscriber (yet/anymore): keep draining
try:
handler(notification)
except Exception:
logger.warning(
"Error dispatching daemon notification %r", notification, exc_info=True
)
finally:
del handler # never hold the subscriber strongly while parked
except Exception:
logger.warning("Daemon notification watcher stopped unexpectedly", exc_info=True)
finally:
self._channel.destroy()
[docs]
@dataclass(frozen=True)
class ClientConfig:
"""Typed configuration for :class:`Client`.
All fields are optional; ``None`` means "use the built-in default".
Example::
client = Client(config=ClientConfig(
cache=CacheConfig(data_bytes=2 * 1024 * 1024 * 1024),
))
"""
cache: Optional[CacheConfig] = None
"""Default cache sizes applied atomically on every ``link_filespace`` call."""
def _to_native_dict(self) -> Dict[str, str]:
"""Translate to the native daemon's flat string dict."""
return {}
@dataclass(frozen=True)
class _InternalClientConfig(ClientConfig):
webservice_url: Optional[str] = None
def _to_native_dict(self) -> Dict[str, str]:
out = super()._to_native_dict()
if self.webservice_url is not None:
out["webservice.url"] = self.webservice_url
return out
[docs]
class Client:
"""LucidLink SDK client.
Manages the connection to LucidLink for one service-account credential.
Each ``Client`` instance is fully independent; multiple clients can run
side-by-side in the same process and operate on different workspaces
concurrently.
Lifecycle::
client = Client()
client.login(credentials)
workspace = client.get_workspace(workspace_id)
# ... use workspace ...
client.close()
Or as a context manager::
with Client() as client:
client.login(credentials)
workspace = client.get_workspace(workspace_id)
# ... use workspace ...
Multi-client (concurrent multi-account)::
client_a = Client(storage=StorageConfig(mode=StorageMode.SANDBOXED))
client_b = Client(storage=StorageConfig(mode=StorageMode.SANDBOXED))
client_a.login(creds_a)
client_b.login(creds_b)
# both fully independent, with separate storage
.. note::
When running multiple concurrent clients, give each one a distinct
``Storage`` so their per-filespace state lives under separate roots.
"""
def __init__(
self,
*,
storage: Optional[StorageConfig] = None,
config: Optional[ClientConfig] = None,
):
"""Construct a client. No I/O is performed until ``login()`` is called.
Args:
storage: Storage configuration. Defaults to ``SANDBOXED`` (temp
directory, cleaned up on close). For concurrent clients,
supply a distinct ``StorageConfig`` per client.
config: Typed client configuration. See :class:`ClientConfig`.
Raises:
ClientError: If the client cannot be initialized.
"""
self._storage = storage or StorageConfig(mode=StorageMode.SANDBOXED)
internal_config = config._to_native_dict() if config else {}
root_str = str(self._storage.get_root_path())
internal_config.setdefault("rootPath", root_str)
internal_config.setdefault("configPath", root_str)
try:
self._native = lucidlink_native.Daemon(
internal_config, cache=config.cache if config else None
)
except Exception as e:
raise ClientError(f"Failed to initialize client: {e}") from e
self._started = False
self._token: Optional[str] = None
self._workspace_info: Optional[WorkspaceInfo] = None
self._workspace: Optional[Workspace] = None
self._notification_watcher: Optional[_NotificationWatcher] = None
@property
def is_logged_in(self) -> bool:
"""Whether ``login()`` has been called successfully and not yet ``close()``-d."""
return self._workspace_info is not None
[docs]
def login(self, credentials: ServiceAccountCredentials) -> None:
"""Authenticate to LucidLink with a service account token.
The token determines which workspace this client operates on.
Args:
credentials: Service account credentials.
Raises:
AuthenticationError: If already logged in with different
credentials, or if authentication fails.
ClientError: If the client cannot be started.
"""
if self.is_logged_in:
if credentials.token == self._token:
return
raise AuthenticationError(
"Client is already logged in. Call close() before logging "
"in with different credentials, or use a new Client."
)
if not self._started:
try:
self._native.start()
self._started = True
self._start_notification_watcher()
except Exception as e:
raise ClientError(f"Failed to start client runtime: {e}") from e
try:
workspace_context = self._native.authenticate(credentials.token)
except Exception as e:
raise AuthenticationError(f"Authentication failed: {e}") from e
self._token = credentials.token
self._workspace_info = WorkspaceInfo(
id=workspace_context["workspace_id"],
name=workspace_context["workspace_name"],
)
self._workspace = Workspace(
native_daemon=self._native,
workspace_id=self._workspace_info.id,
workspace_name=self._workspace_info.name,
notification_watcher=self._notification_watcher,
)
[docs]
def list_workspaces(self) -> List[WorkspaceInfo]:
"""List workspaces this client's credentials grant access to.
Returns:
A list of :class:`WorkspaceInfo` descriptors, one per accessible
workspace. Pass an id to :meth:`get_workspace` to obtain an
operable :class:`Workspace` handle.
Raises:
AuthenticationError: If the client is not logged in.
"""
self._require_logged_in()
return [self._workspace_info]
[docs]
def get_workspace(self, id: str) -> Workspace:
"""Return the workspace with the given id.
Args:
id: The workspace id (from ``list_workspaces()``).
Returns:
The live :class:`Workspace` for filespace operations.
Raises:
AuthenticationError: If the client is not logged in.
ValueError: If ``id`` does not match any accessible workspace.
"""
self._require_logged_in()
if id != self._workspace_info.id:
raise ValueError(
f"Unknown workspace id: {id!r}. "
f"Available: {self._workspace_info.id!r}."
)
return self._workspace
[docs]
def close(self) -> None:
"""Tear down the client: unlink all filespaces and release resources.
Safe to call multiple times. After ``close()``, the client must not
be reused — create a new ``Client`` for further work.
"""
if self._workspace is not None:
try:
self._workspace.stop()
except Exception:
pass
self._workspace = None
if self._started:
try:
self._native.stop()
except Exception:
pass
self._started = False
watcher = self._notification_watcher
self._notification_watcher = None
if watcher is not None:
# The native stop above ended the notification channel, so the watcher is
# already exiting — the join is only synchronization. If it still
# fails to exit in time: it is a daemon thread and sole owner of
# its channel handle, so leaving it behind is safe.
watcher.join(timeout=5.0)
if watcher.is_alive():
logger.warning(
"Daemon notification watcher did not exit within 5s; "
"leaving the daemon thread to finish on its own"
)
self._token = None
self._workspace_info = None
if self._storage.should_cleanup():
root = self._storage.get_root_path()
if root.exists():
shutil.rmtree(root, ignore_errors=True)
def __enter__(self) -> "Client":
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
self.close()
def __repr__(self) -> str:
if self._workspace_info is not None:
return f"Client(workspace='{self._workspace_info.name}')"
return "Client(logged_out)"
def _require_logged_in(self) -> None:
if not self.is_logged_in:
raise AuthenticationError(
"Client is not logged in. Call login() first."
)
def _start_notification_watcher(self) -> None:
"""Start the daemon-notification watcher (idempotent).
Its thread parks off-GIL in the native notification channel so push
notifications (filespace internal errors) are applied the moment they
happen, instead of on the next SDK call. The ``Workspace`` created at login subscribes
itself as the handler.
"""
if self._notification_watcher is not None and self._notification_watcher.is_alive():
return
self._notification_watcher = _NotificationWatcher(self._native)