"""LucidLink Connect - external file management.
Provides the ability to attach existing S3 datasets to a LucidLink filespace
as read-only external files (1:1 file-to-S3-object mapping).
Also supports linking publicly accessible HTTP URLs directly as external files,
without the need for an S3 data store.
"""
from typing import List, Optional
from .connect_models import (
DataStoreCredentials,
DataStoreInfo,
LinkedFile,
LinkedFilesResult,
S3DataStoreConfig,
)
from .connect_validation import (
normalize_data_store_name,
normalize_s3_data_store_config,
validate_access_key,
validate_data_store_count,
validate_data_store_name,
validate_link_file_metadata,
validate_s3_data_store,
validate_secret_key,
)
[docs]
class ConnectManager:
"""Manages LucidLink Connect data stores and external file linking.
Obtained via ``filespace.connect`` after linking to a filespace.
Not intended to be constructed directly.
Example:
.. code-block:: python
connect = filespace.connect
connect.add_data_store("my-store", S3DataStoreConfig(
access_key="AKIA...",
secret_key="...",
bucket_name="my-bucket",
region="us-east-1",
))
connect.link_file("/data/file.csv", "my-store", "path/to/file.csv")
"""
def __init__(self, native_connect):
"""Initialize ConnectManager with native Connect wrapper.
Args:
native_connect: Native Connect wrapper (internal use)
:meta private-args: native_connect
"""
self._native = native_connect
# Data Store Management
[docs]
def add_data_store(self, name: str, config: S3DataStoreConfig) -> DataStoreInfo:
"""Add a new S3 data store.
The new data store is registered in this client's local metadata and
appears in ``list_data_stores()`` immediately. It is not propagated
to LucidLink services until the change is synced: call
``filespace.sync_all()`` before ``list_external_files()``,
``count_external_files()``, ``remove_data_store()``, or
``rekey_data_store()`` — those are served by LucidLink services and
fail while the data store has not reached them yet.
Args:
name: Unique name for the data store
config: S3 data store configuration
Returns:
DataStoreInfo for the newly created data store
Raises:
RuntimeError: If data store creation fails
ValueError: If name is empty or config is invalid
"""
self._ensure_available()
name = normalize_data_store_name(name)
config = normalize_s3_data_store_config(config)
validate_data_store_name(name)
validate_s3_data_store(config)
validate_data_store_count(len(self.list_data_stores()))
result = self._native.add_data_store(
name,
config.access_key,
config.secret_key,
config.bucket_name,
config.region,
config.endpoint,
config.url_expiration_minutes,
config.use_virtual_addressing,
)
return DataStoreInfo.from_dict(result)
[docs]
def add_http_link_data_store(self, name: str) -> DataStoreInfo:
"""Add a new HTTP-link data store. Available on filespace format 3.8+
Args:
name: Unique name for the data store
Returns:
DataStoreInfo for the newly created data store
Raises:
RuntimeError: If data store creation fails
ValueError: If name is empty or invalid
"""
self._ensure_available()
name = normalize_data_store_name(name)
validate_data_store_name(name)
validate_data_store_count(len(self.list_data_stores()))
result = self._native.add_http_link_data_store(name)
return DataStoreInfo.from_dict(result)
[docs]
def remove_data_store(self, name: str) -> None:
"""Remove a data store by name.
Args:
name: Name of the data store to remove
Raises:
RuntimeError: If data store not found or removal fails
"""
self._ensure_available()
name = normalize_data_store_name(name)
self._native.remove_data_store(name)
[docs]
def list_data_stores(self) -> List[DataStoreInfo]:
"""List all registered data stores.
Served from this client's local metadata: data stores added or
removed on this client are visible here immediately, without
``sync_all()``.
Returns:
List of DataStoreInfo objects (secret_key will be empty)
Raises:
RuntimeError: If operation fails
"""
self._ensure_available()
stores = self._native.list_data_stores()
return [DataStoreInfo.from_dict(store) for store in stores]
[docs]
def get_data_store(self, name: str) -> Optional[DataStoreInfo]:
"""Get a data store by name.
Args:
name: Name of the data store
Returns:
DataStoreInfo if found (includes decrypted secret_key), None otherwise
"""
self._ensure_available()
name = normalize_data_store_name(name)
store = self._native.get_data_store(name)
if store is None:
return None
return DataStoreInfo.from_dict(store)
[docs]
def rekey_data_store(
self,
name: str,
credentials: DataStoreCredentials | None = None,
*,
new_access_key: str | None = None,
new_secret_key: str | None = None,
) -> None:
"""Rotate credentials for a data store.
Args:
name: Name of the data store
credentials: Typed credentials object
new_access_key: New S3 access key ID
new_secret_key: New S3 secret access key
Raises:
RuntimeError: If data store not found or rekey fails
ValueError: If credentials are missing or empty
Note:
The rotation is asynchronous: this call initiates it, and
LucidLink services then migrate the store's linked files to the
new credentials in the background. ``DataStoreInfo.rekey_state``
reports ``IN_PROGRESS`` until the migration completes; while it
is in progress, ``list_external_files()`` /
``count_external_files()`` results may be incomplete or
unstable. The data store must already be synced
(``filespace.sync_all()``) before it can be rekeyed.
"""
self._ensure_available()
name = normalize_data_store_name(name)
if credentials is not None:
access_key = credentials.access_key
secret_key = credentials.secret_key
elif new_access_key is not None and new_secret_key is not None:
access_key = new_access_key
secret_key = new_secret_key
else:
raise ValueError(
"Provide either a credentials object or both new_access_key and new_secret_key"
)
validate_access_key(access_key)
validate_secret_key(secret_key)
self._native.rekey_data_store(name, access_key, secret_key)
# External File Operations
[docs]
def link_file(
self,
file_path: str,
data_store_name: str,
object_id: str,
size: int | None = None,
checksum: str = "",
) -> None:
"""Link an S3 object as an external file in the filespace.
Creates a read-only file entry that maps to an S3 object via presigned URL.
The new link is written to this client's local metadata; it is not
reflected by ``list_external_files()`` / ``count_external_files()``
— which are served by LucidLink services — until the change is
synced — call ``filespace.sync_all()`` before listing.
Args:
file_path: Path in the filespace for the new file
data_store_name: Name of the data store containing the object
object_id: S3 object key
size: Optional object size in bytes. When provided together with
checksum, skips the S3 HeadObject call (much faster for bulk ops).
checksum: Optional object ETag/checksum. When provided together with
size, skips the S3 HeadObject call.
Raises:
RuntimeError: If data store not found or link fails
ValueError: If only one of size/checksum is provided
"""
self._ensure_available()
data_store_name = normalize_data_store_name(data_store_name)
validate_link_file_metadata(size, checksum)
self._native.link_file(file_path, data_store_name, object_id, size, checksum)
[docs]
def link_http_file(
self,
file_path: str,
url: str,
*,
data_store_name: str | None = None,
size: int | None = None,
) -> None:
"""Link an HTTP URL as an external file in the filespace.
Creates a read-only file entry that maps to content at the given URL.
Args:
file_path: Path in the filespace for the new file
url: HTTP URL of the file content
data_store_name: Name of an http link data store
Required on filespace format 3.8+, should be None otherwise
size: Optional file size in bytes. When provided, skips the access
check and size validation (useful for bulk operations where
size is known in advance).
Raises:
ValueError: If size is negative
RuntimeError: If the URL is inaccessible or expired, the data
store is not found, or the link fails
"""
if size is not None and size < 0:
raise ValueError("size must be a non-negative integer")
self._ensure_available()
if data_store_name is not None:
data_store_name = normalize_data_store_name(data_store_name)
self._native.link_http_file(file_path, url, data_store_name, size)
[docs]
def update_http_file_url(self, file_path: str, new_url: str, *, size: int | None = None) -> None:
"""Update the URL of an existing HTTP link file.
Args:
file_path: Path of the HTTP link file in the filespace
new_url: New HTTP URL for the file content
size: Optional file size in bytes. When provided, skips the access
check and size validation (useful for bulk operations where
size is known in advance).
Raises:
ValueError: If size is negative
RuntimeError: If the file is not found or is not an HTTP link file
"""
if size is not None and size < 0:
raise ValueError("size must be a non-negative integer")
self._ensure_available()
self._native.update_http_file_url(file_path, new_url, size)
[docs]
def unlink_file(self, file_path: str) -> None:
"""Remove an external file link from the filespace.
The removal is written to this client's local metadata;
``list_external_files()`` / ``count_external_files()`` — which are
served by LucidLink services — keep reporting the file until the
change is synced — call ``filespace.sync_all()`` after unlinking.
Args:
file_path: Path of the external file to remove
Raises:
RuntimeError: If file not found or unlink fails
"""
self._ensure_available()
self._native.unlink_file(file_path)
[docs]
def list_external_files(
self, data_store_name: str, limit: int = 100, cursor: str = ""
) -> LinkedFilesResult:
"""List external files linked to a data store.
The listing is served by LucidLink services: the index of linked
files is maintained there, not in this client's local metadata.
Locally made changes — a newly added data store, newly linked or
unlinked files — are not reflected until they are synced to LucidLink
services. Call ``filespace.sync_all()`` after ``add_data_store()``,
``link_file()``, or ``unlink_file()`` before listing.
Args:
data_store_name: Name of the data store
limit: Maximum number of results (default: 100)
cursor: Opaque pagination token from a previous result (default: start)
Returns:
LinkedFilesResult with the linked entries (path, entry ID, inode
ID per file), pagination flag, and cursor
Raises:
KeyError: If the data store is not known to LucidLink services —
including a data store added on this client but not yet
synced
RuntimeError: If the operation fails
Note:
While a credential rotation started by ``rekey_data_store()`` is
in progress (``DataStoreInfo.rekey_state`` is ``IN_PROGRESS``),
results — in particular paginated ones — may be incomplete or
unstable, because LucidLink services migrate linked files to the
new key in the background. Prefer listing after the rotation
completes.
"""
self._ensure_available()
data_store_name = normalize_data_store_name(data_store_name)
result = self._native.list_external_files(data_store_name, limit, cursor)
return LinkedFilesResult(
file_paths=result["file_paths"],
file_ids=result["entry_ids_external"],
entries=[LinkedFile(**e) for e in result["entries"]],
has_more=result["has_more"],
cursor=result["cursor"],
)
[docs]
def count_external_files(self, data_store_name: str) -> int:
"""Count external files linked to a data store (no path resolution).
Much faster than iterating list_external_files() when only the count
is needed, as it skips per-file path lookups in the metadata store.
Like ``list_external_files()``, the count is served by LucidLink
services: call ``filespace.sync_all()`` first so locally made
changes are reflected. The same caveat about an in-progress
credential rotation applies.
Args:
data_store_name: Name of the data store
Returns:
Number of linked files
Raises:
KeyError: If the data store is not known to LucidLink services —
including a data store added on this client but not yet
synced
RuntimeError: If the operation fails
"""
self._ensure_available()
data_store_name = normalize_data_store_name(data_store_name)
return self._native.count_external_files(data_store_name)
[docs]
def are_data_stores_available(self) -> bool:
"""Check if data stores can be managed on this filespace.
Verifies both that the filespace version supports external files (V9+)
and that the feature has been explicitly configured.
Returns:
True if data stores are available, False otherwise
"""
return self._native.are_data_stores_available()
def _ensure_available(self) -> None:
if not self.are_data_stores_available():
raise RuntimeError(
"Connect (external files) is not available. "
"Ensure the filespace is V9+ and the feature is configured."
)
def __repr__(self) -> str:
return "ConnectManager()"