Source code for shapelib.core.cloud.cloud_path

"""Value Object for parsing and representing cloud/local file paths."""

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Final
from urllib.parse import urlparse




@dataclass(frozen=True)
class CloudPath:
    """
    Immutable Value Object representing a file path (local or remote).

    This class parses URLs and local paths to extract scheme, host, path,
    and file extension. It handles query strings in URLs (e.g., AWS presigned
    URLs) correctly by extracting the extension from the path component only.

    Parameters
    ----------
    raw_path : str
        The original path or URL string.

    Attributes
    ----------
    raw_path : str
        The original path or URL as provided.
    scheme : str
        The URL scheme (e.g., 'https', 's3', 'gs', 'file', or empty for local).
    host : str
        The host/bucket name for remote URLs, empty for local paths.
    path : str
        The path component (without query string).
    extension : str
        The file extension (e.g., '.fgb', '.shp').
    query : str
        The query string component (for presigned URLs).

    Examples
    --------
    Parse a presigned S3 URL:

    >>> cloud_path = CloudPath.from_string(
    ...     "https://bucket.s3.amazonaws.com/file.fgb?X-Amz-Signature=..."
    ... )
    >>> cloud_path.extension
    '.fgb'
    >>> cloud_path.is_remote
    True

    Parse a local path:

    >>> local_path = CloudPath.from_string("/home/user/data.shp")
    >>> local_path.is_remote
    False

    Big-O
    -----
    O(n) where n is the length of the path string.
    """

    raw_path: str
    scheme: str
    host: str
    path: str
    extension: str
    query: str

    _REMOTE_SCHEMES: Final[frozenset[str]] = frozenset({"http", "https", "s3", "gs", "az"})



    @classmethod
    def from_string(cls, path: str | Path) -> CloudPath:
        """
        Create a CloudPath from a string or Path object.

        Parameters
        ----------
        path : str | Path
            The path or URL to parse.

        Returns
        -------
        CloudPath
            A new CloudPath instance with parsed components.

        Raises
        ------
        ValueError
            If the path is empty or None.
        """
        if not path:
            raise ValueError("Path cannot be empty or None.")

        raw_path = str(path)
        parsed = urlparse(raw_path)

        scheme = parsed.scheme.lower()
        host = parsed.netloc
        url_path = parsed.path
        query = parsed.query

        extension = cls._extract_extension(url_path, raw_path)

        return cls(
            raw_path=raw_path,
            scheme=scheme,
            host=host,
            path=url_path,
            extension=extension,
            query=query,
        )


    @classmethod
    def _extract_extension(cls, url_path: str, raw_path: str) -> str:
        """Extract file extension from path, handling URLs correctly."""
        if url_path:
            return Path(url_path).suffix.lower()
        return Path(raw_path).suffix.lower()

    @property
    def is_remote(self) -> bool:
        """Check if this path represents a remote location."""
        return self.scheme in self._REMOTE_SCHEMES

    @property
    def is_s3_protocol(self) -> bool:
        """Check if this path uses the S3 protocol (s3://)."""
        return self.scheme == "s3"

    @property
    def is_https(self) -> bool:
        """Check if this path uses HTTP/HTTPS protocol."""
        return self.scheme in {"http", "https"}

    @property
    def full_url(self) -> str:
        """Reconstruct the full URL including query string."""
        if self.query:
            return f"{self.scheme}://{self.host}{self.path}?{self.query}"
        if self.is_remote:
            return f"{self.scheme}://{self.host}{self.path}"
        return self.raw_path

    @property
    def s3_bucket(self) -> str | None:
        """Extract S3 bucket name from path."""
        if self.is_s3_protocol:
            return self.host
        if self.is_https and ".s3." in self.host:
            return self.host.split(".s3.")[0]
        return None

    @property
    def s3_key(self) -> str | None:
        """Extract S3 object key from path."""
        if self.is_s3_protocol or (self.is_https and ".s3." in self.host):
            return self.path.lstrip("/")
        return None

    @property
    def vsicurl_path(self) -> str:
        """
        Build the GDAL virtual filesystem path for streaming.

        Returns
        -------
        str
            The /vsicurl/ or /vsis3/ path for GDAL.
        """
        if self.is_s3_protocol:
            return f"/vsis3/{self.s3_bucket}/{self.s3_key}"
        return f"/vsicurl/{self.full_url}"

    def __str__(self) -> str:
        """Return the original raw path string."""
        return self.raw_path