""" Provider-neutral cloud resource capability contract for DocuVault Phase 12. Defines the stable vocabulary (action keys, capability states, reason codes), immutable normalized value types (CloudCapability, CloudResource, CloudListing), and the abstract CloudResourceAdapter read-only interface. Design decisions (12-CONTEXT.md D-06 through D-10): - Structurally unsupported actions remain visible but greyed out (D-06). - Capability discovery must never mutate provider content (D-08). - Permanent vs. temporary limitations have distinct visual states (D-09). - No mutation methods in Phase 12 contract; mutations added in Phase 13. Credentials are never fields on any value type. The interface contains no put, delete, rename, move, or create_folder methods. """ from __future__ import annotations import uuid from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime from typing import Optional # ── Action keys ─────────────────────────────────────────────────────────────── ACTIONS = frozenset({ "browse", "open", "preview", "upload", "create_folder", "rename", "move", "delete", "change_tracking", }) # ── Capability states ───────────────────────────────────────────────────────── STATE_SUPPORTED = "supported" STATE_UNSUPPORTED = "unsupported" STATE_TEMPORARILY_UNAVAILABLE = "temporarily_unavailable" CAPABILITY_STATES = frozenset({ STATE_SUPPORTED, STATE_UNSUPPORTED, STATE_TEMPORARILY_UNAVAILABLE, }) # ── Stable reason codes ─────────────────────────────────────────────────────── REASON_PROVIDER_UNSUPPORTED = "provider_unsupported" REASON_INSUFFICIENT_SCOPE = "insufficient_scope" REASON_READ_ONLY = "read_only" REASON_REAUTH_REQUIRED = "reauth_required" REASON_OFFLINE = "offline" REASON_ITEM_RESTRICTED = "item_restricted" KNOWN_REASONS = frozenset({ REASON_PROVIDER_UNSUPPORTED, REASON_INSUFFICIENT_SCOPE, REASON_READ_ONLY, REASON_REAUTH_REQUIRED, REASON_OFFLINE, REASON_ITEM_RESTRICTED, }) # ── Normalized value types ──────────────────────────────────────────────────── @dataclass(frozen=True) class CloudCapability: """Normalized capability descriptor for a single action. state: one of the three CAPABILITY_STATES constants. reason: one of the KNOWN_REASONS codes when state != supported; None otherwise. message: controlled adapter output — safe for display; never raw provider text. Invariants enforced at construction: - state must be a known capability state. - reason and message are only present when state is not supported. - reason, when present, must be a known reason code. """ action: str state: str reason: Optional[str] = None message: Optional[str] = None def __post_init__(self) -> None: if self.state not in CAPABILITY_STATES: raise ValueError( f"Unknown capability state {self.state!r}; " f"must be one of {sorted(CAPABILITY_STATES)}" ) if self.action not in ACTIONS: raise ValueError( f"Unknown action {self.action!r}; must be one of {sorted(ACTIONS)}" ) if self.state == STATE_SUPPORTED: if self.reason is not None or self.message is not None: raise ValueError( "reason and message must be None when state is 'supported'" ) else: if self.reason is not None and self.reason not in KNOWN_REASONS: raise ValueError( f"Unknown reason code {self.reason!r}; " f"must be one of {sorted(KNOWN_REASONS)}" ) @dataclass(frozen=True) class CloudResource: """Normalized cloud item metadata — provider-independent representation. id: stable DocuVault UUID for this item; survives provider rename/move. provider_item_id: opaque provider-assigned identifier (e.g. Drive file ID). connection_id: UUID of the CloudConnection that owns this item. user_id: UUID of the owning user (security boundary). name: display name as reported by the provider. kind: "file" or "folder". parent_ref: opaque provider reference to the parent folder; None at root. content_type: MIME type for files; None for folders. size: provider-reported size in bytes; None if not reported or for folders. modified_at: provider-reported modification time; None if not available. etag: provider etag or version string for change detection; None if absent. capabilities: per-item capability overrides merged with connection defaults. """ id: uuid.UUID provider_item_id: str connection_id: uuid.UUID user_id: uuid.UUID name: str kind: str # "file" | "folder" parent_ref: Optional[str] = None content_type: Optional[str] = None size: Optional[int] = None modified_at: Optional[datetime] = None etag: Optional[str] = None capabilities: dict[str, CloudCapability] = field(default_factory=dict) def __post_init__(self) -> None: if self.kind not in ("file", "folder"): raise ValueError(f"kind must be 'file' or 'folder', got {self.kind!r}") for action, cap in self.capabilities.items(): if not isinstance(cap, CloudCapability): raise TypeError( f"capabilities[{action!r}] must be a CloudCapability instance" ) @dataclass(frozen=True) class CloudListing: """Result of listing a folder from a provider. items: normalized CloudResource children of the listed folder. complete: True when the listing represents the full authoritative contents (safe to mark unseen children deleted on reconciliation). False when the listing is partial, paginated, or failed — retained rows must NOT be marked deleted. next_page_token: opaque token for continuation; None when exhausted. """ items: tuple[CloudResource, ...] complete: bool = True next_page_token: Optional[str] = None def __post_init__(self) -> None: # Coerce list to tuple for immutability object.__setattr__(self, "items", tuple(self.items)) # ── Abstract adapter ────────────────────────────────────────────────────────── class CloudResourceAdapter(ABC): """Read-only cloud resource adapter contract for Phase 12. Phase 12 interface is intentionally limited to browse and capability discovery. No method here creates, renames, moves, or deletes provider content. Mutation methods are added in Phase 13 on a separate subclass interface. Credentials are passed at construction time (or via dependency injection) and are never exposed as fields or return values. """ @abstractmethod async def list_folder( self, connection_id: uuid.UUID, user_id: uuid.UUID, parent_ref: Optional[str] = None, page_token: Optional[str] = None, ) -> CloudListing: """List the direct children of a folder. parent_ref=None lists the connection root. Returns a CloudListing with complete=False on partial or failed fetches. Raises CloudConnectionError (from storage.exceptions) on auth failures. Never mutates provider content as a side effect. """ ... @abstractmethod async def get_capabilities( self, connection_id: uuid.UUID, user_id: uuid.UUID, ) -> dict[str, CloudCapability]: """Return connection-level capabilities for all defined ACTIONS. The returned dict must contain an entry for every action in ACTIONS. Probing must not create, rename, move, or delete any provider content. """ ... def merge_item_capabilities( self, connection_caps: dict[str, CloudCapability], item_caps: dict[str, CloudCapability], ) -> dict[str, CloudCapability]: """Merge per-item overrides onto connection defaults. Item capabilities take precedence over connection capabilities. Actions absent from item_caps fall back to connection_caps. The result always contains every action in ACTIONS. """ merged: dict[str, CloudCapability] = {} for action in ACTIONS: if action in item_caps: merged[action] = item_caps[action] elif action in connection_caps: merged[action] = connection_caps[action] else: # Safe default: unsupported with no reason merged[action] = CloudCapability( action=action, state=STATE_UNSUPPORTED, reason=REASON_PROVIDER_UNSUPPORTED, message="This action is not supported by the provider.", ) return merged