"""
Workspace context after successful authentication.
"""
from typing import List, Optional
from .filespace import Filespace
from .filespace_models import FilespaceInfo, SyncMode
[docs]
class Workspace:
"""
Workspace context after successful authentication.
Provides access to filespace operations within the authenticated workspace.
Returned by ``Daemon.authenticate()``.
Example:
.. code-block:: python
credentials = ServiceAccountCredentials(token)
workspace = daemon.authenticate(credentials)
print(workspace.id, workspace.name)
"""
def __init__(self, native_daemon, workspace_id: str, workspace_name: str):
"""
Initialize workspace context.
Args:
native_daemon: Native daemon wrapper (internal use)
workspace_id: Workspace ID
workspace_name: Workspace name
Note: This constructor is called internally by ``Daemon.authenticate()``.
Users should not construct ``Workspace`` objects directly.
"""
self._native_daemon = native_daemon
self._id = workspace_id
self._name = workspace_name
self._active_filespace = None
@property
def id(self) -> str:
"""Get the workspace ID."""
return self._id
@property
def name(self) -> str:
"""Get the workspace name."""
return self._name
[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 WebService unreachable
AuthenticationError: If access token expired
RuntimeError: If daemon 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.
Args:
name: Filespace name (either ``name`` or ``id`` required)
id: Filespace ID (either ``name`` or ``id`` required)
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 daemon not running or not authenticated
Example:
.. code-block:: python
# Link by name (user-friendly)
fs = workspace.link_filespace(name="production-data")
# Link by ID (efficient if you have the UUID)
fs = workspace.link_filespace(id="fs-uuid-12345")
# Using as context manager (auto sync + unlink on exit)
with workspace.link_filespace(name="data") as fs:
fs.fs.write_file("/file.txt", b"data")
# sync_all() + unlink() called automatically
# Disable auto-sync
fs = workspace.link_filespace(name="data", sync_mode=SyncMode.SYNC_NONE)
"""
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'")
# Unlink previous filespace if re-linking
if self._active_filespace is not None:
try:
self._active_filespace.unlink()
except Exception:
pass
self._active_filespace = None
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
)
filespace = Filespace(
native_fs=native_fs,
native_daemon=self._native_daemon,
workspace_id=self._id,
workspace_name=self._name,
id=native_fs.get_id() or "",
full_name=native_fs.get_name() or "",
sync_mode=sync_mode,
)
self._active_filespace = filespace
return filespace
[docs]
def stop(self) -> None:
"""Stop workspace — unlinks active filespace if any.
If ``sync_mode`` is ``SYNC_ALL``, calls ``sync_all()`` before unlinking.
Safe to call multiple times.
"""
if self._active_filespace is not None:
try:
self._active_filespace.unlink()
except Exception:
pass
self._active_filespace = None
def __repr__(self) -> str:
return f"Workspace(id='{self.id}', name='{self.name}')"