"""Data models for LucidLink Connect (external files)."""
import enum
from dataclasses import dataclass, field
from typing import List
[docs]
class DataStoreKind(str, enum.Enum):
"""Type of external data store."""
S3 = "S3DataStore"
"""Amazon S3 or S3-compatible object storage."""
[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
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."""
[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 (currently always S3)."""
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")),
}
)
[docs]
@dataclass
class LinkedFilesResult:
"""Result from listing external files linked to a data store."""
file_paths: List[str] = field(default_factory=list)
"""List of file paths."""
file_ids: List[int] = field(default_factory=list)
"""List of integer file IDs (parallel to ``file_paths``)."""
has_more: bool = False
"""Whether more results are available (pagination)."""
cursor: str = ""
"""Opaque pagination token; pass to the next call to continue."""