API Reference¶

@dataclass(frozen=True)
class Artifact:
    """A strict, immutable schema-v1 materialization envelope."""

    uri: str
    media_type: str
    size: int
    content_digest: Digest
    blob_digest: Digest
    provenance: ArtifactProvenance
    metadata: Mapping[str, object] = field(default_factory=dict)
    extensions: Mapping[str, object] = field(default_factory=dict)

    def __post_init__(self) -> None:
        _public_uri(self.uri, "uri")
        if not isinstance(self.media_type, str) or not self.media_type:
            raise _contract_error("media_type")
        if type(self.size) is not int or self.size < 0:
            raise _contract_error("size")
        if not isinstance(self.content_digest, Digest) or not isinstance(self.blob_digest, Digest):
            raise _contract_error("digest")
        if not isinstance(self.provenance, ArtifactProvenance):
            raise _contract_error("provenance")
        metadata = _freeze_json(self.metadata, "metadata")
        if not isinstance(metadata, Mapping):
            raise _contract_error("metadata")
        object.__setattr__(self, "metadata", metadata)
        object.__setattr__(self, "extensions", _freeze_extensions(self.extensions))

    def to_dict(self) -> dict[str, object]:
        """Return a fresh JSON-safe Artifact envelope."""
        return {
            "schema_version": 1,
            "kind": "artifact",
            "uri": self.uri,
            "media_type": self.media_type,
            "size": self.size,
            "content_digest": self.content_digest.to_dict(),
            "blob_digest": self.blob_digest.to_dict(),
            "provenance": self.provenance.to_dict(),
            "metadata": _thaw_json(self.metadata),
            "extensions": _thaw_json(self.extensions),
        }

    @classmethod
    def from_dict(cls, value: object) -> Artifact:
        """Decode one strict schema-v1 Artifact envelope."""
        data = _object_dict(value, "artifact")
        if set(data) != _ARTIFACT_KEYS:
            raise _contract_error("artifact")
        if data["schema_version"] != 1 or type(data["schema_version"]) is not int or data["kind"] != "artifact":
            raise _contract_error("artifact")
        uri = data["uri"]
        media_type = data["media_type"]
        size = data["size"]
        metadata = data["metadata"]
        extensions = data["extensions"]
        if not isinstance(uri, str) or not isinstance(media_type, str) or type(size) is not int:
            raise _contract_error("artifact")
        return cls(
            uri=uri,
            media_type=media_type,
            size=size,
            content_digest=Digest.from_dict(data["content_digest"]),
            blob_digest=Digest.from_dict(data["blob_digest"]),
            provenance=ArtifactProvenance.from_dict(data["provenance"]),
            metadata=_object_dict(metadata, "metadata"),
            extensions=_object_dict(extensions, "extensions"),
        )

@dataclass(frozen=True)
class ArtifactProvenance:
    """Typed provenance for one materialized Artifact."""

    source_locator: ResourceLocator
    resource_identity: str
    created_at: datetime
    materialization_mode: str
    transforms: tuple[str, ...] = ()

    def __post_init__(self) -> None:
        if not hasattr(self.source_locator, "to_dict") or not _is_sha256(self.resource_identity):
            raise _contract_error("provenance")
        if not isinstance(self.created_at, datetime) or self.created_at.tzinfo is None:
            raise _contract_error("provenance.created_at")
        if self.materialization_mode not in {"parquet", "raw"}:
            raise _contract_error("provenance.materialization_mode")
        if not isinstance(self.transforms, tuple) or not all(isinstance(value, str) for value in self.transforms):
            raise _contract_error("provenance.transforms")

    def to_dict(self) -> dict[str, object]:
        """Return a fresh JSON-safe provenance envelope."""
        created_at = self.created_at.astimezone(UTC).isoformat().replace("+00:00", "Z")
        return {
            "source_locator": self.source_locator.to_dict(),
            "resource_identity": self.resource_identity,
            "created_at": created_at,
            "materialization_mode": self.materialization_mode,
            "transforms": list(self.transforms),
        }

    @classmethod
    def from_dict(cls, value: object) -> ArtifactProvenance:
        """Decode one strict provenance envelope."""
        data = _object_dict(value, "provenance")
        if set(data) != _PROVENANCE_KEYS:
            raise _contract_error("provenance")
        source_locator = data["source_locator"]
        resource_identity = data["resource_identity"]
        created_at = data["created_at"]
        materialization_mode = data["materialization_mode"]
        transforms = data["transforms"]
        if (
            not isinstance(resource_identity, str)
            or not isinstance(created_at, str)
            or not isinstance(materialization_mode, str)
            or not isinstance(transforms, list)
            or not all(isinstance(transform, str) for transform in transforms)
        ):
            raise _contract_error("provenance")
        try:
            parsed_created_at = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
        except ValueError as exc:
            raise _contract_error("provenance.created_at") from exc
        if parsed_created_at.tzinfo is None:
            raise _contract_error("provenance.created_at")
        from datasluice.application import resource_locator_from_dict

        return cls(
            source_locator=resource_locator_from_dict(_object_dict(source_locator, "provenance.source_locator")),
            resource_identity=resource_identity,
            created_at=parsed_created_at,
            materialization_mode=materialization_mode,
            transforms=cast(tuple[str, ...], tuple(transforms)),
        )

class CatalogError(DataSluiceError):
    """Base normalized error with a portable safe next action."""

    def __init__(
        self,
        message: str,
        *,
        operation: str,
        platform: CatalogPlatform | str,
        capability_state: str | None = None,
        safe_action: str,
    ) -> None:
        if not isinstance(message, str) or not message:
            raise ValueError("Catalog error messages must be non-empty strings.")
        if not isinstance(operation, str) or not operation:
            raise ValueError("Catalog error operations must be non-empty strings.")
        if not isinstance(safe_action, str) or not safe_action:
            raise ValueError("Catalog errors require a safe next action.")
        super().__init__(message)
        self.operation = operation
        self.platform = _platform_value(platform)
        self.capability_state = capability_state
        self.safe_action = safe_action

@dataclass(frozen=True)
class CatalogId:
    """A platform- and resource-kind-scoped opaque catalog identifier."""

    platform: CatalogPlatform
    resource_kind: ResourceKind
    value: str

    def __post_init__(self) -> None:
        if not isinstance(self.platform, CatalogPlatform):
            raise _contract_error("catalog_id.platform")
        if not isinstance(self.resource_kind, ResourceKind):
            raise _contract_error("catalog_id.resource_kind")
        if not isinstance(self.value, str) or not self.value:
            raise _contract_error("catalog_id.value")

    def to_dict(self) -> dict[str, object]:
        """Return a fresh JSON-safe catalog identifier envelope."""
        return {
            "schema_version": 1,
            "kind": "catalog_id",
            "platform": self.platform.value,
            "resource_kind": self.resource_kind.value,
            "value": self.value,
        }

    @classmethod
    def from_dict(cls, value: object) -> CatalogId:
        """Decode one strict schema-v1 catalog identifier envelope."""
        if not isinstance(value, dict) or set(value) != _ID_KEYS:
            raise _contract_error("catalog_id")
        if value["schema_version"] != 1 or type(value["schema_version"]) is not int or value["kind"] != "catalog_id":
            raise _contract_error("catalog_id")
        platform = value["platform"]
        resource_kind = value["resource_kind"]
        native_value = value["value"]
        if not isinstance(platform, str) or not isinstance(resource_kind, str) or not isinstance(native_value, str):
            raise _contract_error("catalog_id")
        return cls(CatalogPlatform(platform), ResourceKind(resource_kind), native_value)

@dataclass(frozen=True)
class CatalogPlatform:
    """An explicit platform identity for one catalog deployment family."""

    value: str

    CKAN: ClassVar[CatalogPlatform]
    UDATA: ClassVar[CatalogPlatform]
    SOCRATA: ClassVar[CatalogPlatform]

    def __post_init__(self) -> None:
        if not isinstance(self.value, str) or _VALUE_RE.fullmatch(self.value) is None:
            raise _contract_error("platform")

    def __str__(self) -> str:
        """Return the JSON platform value."""
        return self.value

class CatalogUnavailableError(CatalogError):
    """Raised when a catalog deployment or circuit is unavailable."""

class ChecksumMismatchError(DownloadError):
    """Raised when a downloaded file's checksum does not match."""

    def __init__(self, message: str, expected: str | None = None, actual: str | None = None) -> None:
        super().__init__(message)
        self.expected = expected
        self.actual = actual

class ConfigError(DataSluiceError):
    """Raised when configuration is invalid or incomplete."""

@dataclass(frozen=True)
class CredentialScope:
    """Host-scoped policy controlling where credentials may be sent.

    Attributes:
        allowed_hosts: Hostnames the credential may be sent to.
        allowed_schemes: URL schemes the credential may travel over.
        send_on_redirect: Whether credentials are retained on redirects to allowed hosts.
    """

    allowed_hosts: tuple[str, ...] = field(default_factory=tuple)
    allowed_schemes: tuple[str, ...] = ("https",)
    send_on_redirect: bool = False

class DataSluice:
    """Canonical public facade for discovery, resource access, and materialization."""

    def __init__(
        self,
        *,
        session: Any | None = None,
        reader: Any | None = None,
        **session_kwargs: Any,
    ) -> None:
        if session is not None and session_kwargs:
            raise DataSluiceError("session= cannot be combined with session configuration")
        self._owns_session_dependencies = session is None
        self._owns_reader = reader is None
        self._session = session if session is not None else DataSluiceSession(**session_kwargs)
        self._reader = reader if reader is not None else DataPlaneResourceReader(transport=self._session._transport)
        self._services = _ApplicationServices(self._session, self._reader)
        self._owned_closeables = self._collect_owned_closeables(session_kwargs)
        self._closed = False

    def open_catalog[T](self, factory: Callable[[CatalogConnectorContext], T], context: CatalogConnectorContext) -> T:
        """Return one explicit caller-selected canonical catalog connector."""
        self._ensure_open()
        return self._session.open_catalog(factory, context)

    def resolve(self, locator: DirectResourceLocator) -> Resource:
        """Resolve one public locator into the canonical Resource model."""
        self._ensure_open()
        return self._services.resolve(locator)

    def open(self, resource: Resource | DirectResourceLocator) -> OpenedResource:
        """Return a lazy, single-use OpenedResource wrapper."""
        self._ensure_open()
        return self._services.open(resource)

    def materialize(
        self,
        resource: Resource | DirectResourceLocator,
        destination_uri: str,
        *,
        mode: str = "parquet",
    ) -> Any:
        """Materialize one Resource or ResourceLocator into an Artifact."""
        return self._services.materialize(resource, destination_uri, mode=mode)

    def download_many(self, resources: list[Resource], destination: str) -> list[dict[str, object]]:
        """Raw bulk-copy resources into a destination directory."""
        self._ensure_open()
        return self._services.download_many(resources, destination)

    def close(self) -> None:
        """Close this facade and any resource wrappers it owns."""
        if self._closed:
            return
        self._closed = True
        first_error: BaseException | None = None
        for closeable in self._owned_closeables:
            try:
                closeable.close()
            except BaseException as exc:
                if first_error is None:
                    first_error = exc
        if first_error is not None:
            raise first_error

    def __enter__(self) -> DataSluice:
        self._ensure_open()
        return self

    def __exit__(self, *exc: Any) -> None:
        self.close()

    def _ensure_open(self) -> None:
        if self._closed:
            raise StreamClosedError("DataSluice is closed")

    def _collect_owned_closeables(self, session_kwargs: Mapping[str, Any]) -> tuple[Any, ...]:
        candidates: list[Any] = []
        if self._owns_reader:
            candidates.append(self._reader)
        if self._owns_session_dependencies:
            if session_kwargs.get("transport") is None:
                candidates.append(self._session._transport)
            if session_kwargs.get("cache") is None:
                candidates.append(self._session._cache)
            if session_kwargs.get("storage") is None:
                candidates.append(self._session.storage)
            if session_kwargs.get("state_store") is None:
                candidates.append(self._session.state_store)
            if session_kwargs.get("plugins") is None:
                candidates.append(self._session.plugins)
        closeables: list[Any] = []
        seen: set[int] = set()
        for candidate in candidates:
            if candidate is None or not hasattr(candidate, "close") or id(candidate) in seen:
                continue
            seen.add(id(candidate))
            closeables.append(candidate)
        return tuple(closeables)

class DataSluiceError(Exception):
    """Base exception for all DataSluice errors."""

@dataclass(frozen=True)
class Dataset:
    """A dataset is a logical grouping of one or more resources.

    Attributes:
        id: Portal-native dataset identifier.
        title: Human-readable dataset title.
        name: Machine-friendly slug or name.
        description: Longer free-text description (may contain Markdown/HTML).
        resources: List of downloadable resources within this dataset.
        organization: Publishing organization, if known.
        license: Default license for resources in this dataset.
        tags: Free-form tags or keywords.
        themes: Categorization themes or groups.
        language: ISO language code(s) for the data.
        created: ISO-8601 creation timestamp.
        modified: ISO-8601 last-modified timestamp.
        metadata_modified: ISO-8601 timestamp of last metadata change.
        url: Canonical URL to the dataset on the portal.
        extra: Portal-native fields not captured above.
    """

    id: str
    title: str | None = None
    name: str | None = None
    description: str | None = None
    resources: list[Resource] = field(default_factory=list)
    organization: Organization | None = None
    license: License | None = None
    tags: list[str] = field(default_factory=list)
    themes: list[str] = field(default_factory=list)
    language: list[str] = field(default_factory=list)
    created: str | None = None
    modified: str | None = None
    metadata_modified: str | None = None
    url: str | None = None
    extra: dict[str, Any] = field(default_factory=dict)

@dataclass(frozen=True)
class DatasetRecord:
    """A normalized immutable dataset record."""

    id: CatalogId
    name: str
    description: str | None = None
    extensions: Mapping[str, object] = field(default_factory=dict)

    def __post_init__(self) -> None:
        if not isinstance(self.id, CatalogId) or self.id.resource_kind != ResourceKind.DATASET:
            raise _contract_error("dataset.id")
        if not isinstance(self.name, str) or not self.name:
            raise _contract_error("dataset.name")
        _optional_string(self.description, "dataset.description")
        object.__setattr__(self, "extensions", _freeze_extensions(self.extensions))

    def to_dict(self) -> dict[str, object]:
        """Return a fresh JSON-safe dataset envelope."""
        return {
            "schema_version": 1,
            "kind": "dataset",
            "id": self.id.to_dict(),
            "name": self.name,
            "description": self.description,
            "extensions": _thaw_json(self.extensions),
        }

    @classmethod
    def from_dict(cls, value: object) -> DatasetRecord:
        """Decode one strict schema-v1 dataset envelope."""
        data = _strict_envelope(
            value,
            "dataset",
            "dataset",
            frozenset({"schema_version", "kind", "id", "name", "description", "extensions"}),
        )
        name = data["name"]
        if not isinstance(name, str):
            raise _contract_error("dataset.name")
        return cls(
            id=CatalogId.from_dict(data["id"]),
            name=name,
            description=_optional_string(data["description"], "dataset.description"),
            extensions=_object_dict(data["extensions"], "dataset.extensions"),
        )

@dataclass(frozen=True)
class DetectionResult:
    """Outcome of portal auto-detection with confidence and evidence.

    Attributes:
        portal_type: Identified portal type, or ``None`` when undetected.
        confidence: Confidence score in the range ``[0.0, 1.0]``.
        evidence: Evidence records supporting the detection.
        extra: Portal-native detection fields not captured above.
    """

    portal_type: str | None
    confidence: float = 0.0
    evidence: Sequence[DetectionEvidence] = field(default_factory=tuple)
    extra: Mapping[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        if not 0.0 <= self.confidence <= 1.0:
            raise ValueError(f"confidence must be in [0.0, 1.0], got {self.confidence}")
        if not isinstance(self.evidence, tuple):
            object.__setattr__(self, "evidence", tuple(self.evidence))
        if not isinstance(self.extra, MappingProxyType):
            object.__setattr__(self, "extra", MappingProxyType(dict(self.extra)))

@dataclass(frozen=True)
class Digest:
    """A structured SHA-256 digest."""

    algorithm: str
    value: str

    def __post_init__(self) -> None:
        if self.algorithm != "sha256" or not _is_sha256(self.value):
            raise _contract_error("digest")

    def to_dict(self) -> dict[str, str]:
        """Return a fresh JSON-safe digest envelope."""
        return {"algorithm": self.algorithm, "value": self.value}

    @classmethod
    def from_dict(cls, value: object) -> Digest:
        """Decode one strict digest envelope."""
        data = _object_dict(value, "digest")
        if set(data) != _DIGEST_KEYS:
            raise _contract_error("digest")
        algorithm = data["algorithm"]
        digest = data["value"]
        if not isinstance(algorithm, str) or not isinstance(digest, str):
            raise _contract_error("digest")
        return cls(algorithm=algorithm, value=digest)