"""Data models for LucidLink Connect (external files)."""
import enum
from dataclasses import dataclass, field
from typing import List
from ._redaction import redacted_dataclass_repr
[docs]
class DataStoreKind(str, enum.Enum):
"""Type of external data store."""
S3 = "S3DataStore"
"""Amazon S3 or S3-compatible object storage."""
HTTP_LINK = "HttpLinkDataStore"
"""Groups files linked via direct HTTP/HTTPS URLs."""
[docs]
class DataStoreRekeyState(str, enum.Enum):
"""Credential rotation state for a data store."""
NO_REKEY = "no_rekey"
"""No credential rotation in progress."""
IN_PROGRESS = "in_progress"
"""Credential rotation is currently in progress."""
[docs]
@dataclass
class S3Credentials:
access_key: str
secret_key: str
kind: DataStoreKind = DataStoreKind.S3
def __repr__(self) -> str:
return redacted_dataclass_repr(self)
DataStoreCredentials = S3Credentials
[docs]
@dataclass
class S3DataStoreConfig:
"""Configuration for an S3-compatible data store."""
access_key: str
"""S3 access key ID."""
secret_key: str
"""S3 secret access key."""
bucket_name: str
"""S3 bucket name."""
region: str
"""AWS region (e.g., ``"us-east-1"``)."""
endpoint: str = ""
"""Custom S3 endpoint URL with scheme, e.g. ``"http://127.0.0.1:9090"``.
Must start with ``"http://"`` or ``"https://"``. When empty, defaults to HTTPS."""
url_expiration_minutes: int = 10080
"""Presigned URL expiration in minutes (7 days)."""
use_virtual_addressing: bool = False
"""Use virtual-hosted-style addressing."""
def __repr__(self) -> str:
return redacted_dataclass_repr(self)
[docs]
@dataclass
class DataStoreInfo:
"""Information about a registered data store."""
name: str
"""Data store name."""
access_key: str = ""
"""S3 access key ID."""
secret_key: str = ""
"""S3 secret access key."""
bucket_name: str = ""
"""S3 bucket name."""
region: str = ""
"""AWS region."""
endpoint: str = ""
"""Full S3 endpoint URL (e.g., ``"http://127.0.0.1:9090"``), empty if using AWS default."""
url_expiration_minutes: int = 0
"""Presigned URL expiration in minutes."""
use_virtual_addressing: bool = False
"""Whether virtual-hosted-style addressing is used."""
kind: DataStoreKind = DataStoreKind.S3
"""Data store type."""
key_id: str = ""
"""Unique key identifier (UUID)."""
rekey_state: DataStoreRekeyState = DataStoreRekeyState.NO_REKEY
"""Current credential rotation state."""
[docs]
@classmethod
def from_dict(cls, store: dict) -> "DataStoreInfo":
return cls(
**{
**store,
"kind": DataStoreKind(store.get("kind", "S3DataStore")),
"rekey_state": DataStoreRekeyState(store.get("rekey_state", "no_rekey")),
}
)
def __repr__(self) -> str:
return redacted_dataclass_repr(self)
[docs]
@dataclass
class LinkedFile:
"""One external file linked from a data store."""
path: str
"""Path of the linked file within the filespace."""
entry_id: str
"""Entry ID. Matches ``DirEntry.entry_id``; hard links to the same file
have distinct entry IDs."""
inode_id: str
"""Inode ID. Matches ``DirEntry.inode_id``; shared by all hard links to
the same file."""
[docs]
@dataclass
class LinkedFilesResult:
"""Result from listing external files linked from a data store."""
file_paths: List[str] = field(default_factory=list)
"""Deprecated. List of file paths; use ``entries`` instead."""
file_ids: List[int] = field(default_factory=list)
"""Deprecated. Numeric (lossy) entry IDs; use ``entries`` instead."""
has_more: bool = False
"""Whether more results are available (pagination)."""
cursor: str = ""
"""Opaque pagination token; pass to the next call to continue."""
entries: List[LinkedFile] = field(default_factory=list)
"""List of linked files, each carrying its path, entry ID, and inode ID."""