"""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 shutil
from dataclasses import dataclass
from typing import Dict, List, Optional
from . import lucidlink_native
from .credentials import ServiceAccountCredentials
from .exceptions import AuthenticationError, ClientError
from .storage import StorageConfig, StorageMode
from .workspace import Workspace
from .workspace_models import WorkspaceInfo
[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(
fs_cache_size_mb=2048,
))
"""
fs_cache_size_mb: Optional[int] = None
"""Per-filespace disk cache size in megabytes. Default: 1024 MB."""
def _to_native_dict(self) -> Dict[str, str]:
"""Translate to the native daemon's flat string dict."""
out: Dict[str, str] = {}
if self.fs_cache_size_mb is not None:
out["fs.cache.size"] = str(self.fs_cache_size_mb)
return out
@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)
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
@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
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,
)
[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
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."
)