Source code for shapelib.core.cloud.providers.https_provider

"""HTTPS cloud storage provider for presigned URLs."""

from __future__ import annotations

from pathlib import Path
from typing import TYPE_CHECKING

import requests

from shapelib.factories.cloud_registry import register_cloud_provider

if TYPE_CHECKING:
    from shapelib.core.cloud.cloud_path import CloudPath




@register_cloud_provider
class HttpsProvider:
    """
    Cloud provider for HTTPS URLs (including presigned URLs).

    Handles file operations using the requests library for HTTP/HTTPS URLs.
    Works with presigned URLs from any cloud provider (AWS, GCS, Azure).

    Examples
    --------
    >>> provider = HttpsProvider()
    >>> provider.supports(CloudPath.from_string("https://example.com/file.fgb"))
    True

    Big-O
    -----
    All operations are O(n) where n is the file size.
    """



    def supports(self, cloud_path: CloudPath) -> bool:
        """
        Check if this provider handles HTTPS URLs.

        Parameters
        ----------
        cloud_path : CloudPath
            The parsed cloud path to check.

        Returns
        -------
        bool
            True if the path uses http:// or https:// protocol.
        """
        return cloud_path.is_https




    def upload(self, source: Path, cloud_path: CloudPath) -> None:
        """
        Upload a local file via HTTPS PUT (presigned URL).

        Parameters
        ----------
        source : Path
            The local file to upload.
        cloud_path : CloudPath
            The HTTPS destination URL (presigned PUT URL).

        Raises
        ------
        FileNotFoundError
            If the source file does not exist.
        ConnectionError
            If the upload fails.
        PermissionError
            If access is denied (403).
        """
        try:
            with open(source, "rb") as file_handle:
                response = requests.put(
                    cloud_path.full_url,
                    data=file_handle,
                    timeout=300,
                )
                response.raise_for_status()
        except requests.exceptions.HTTPError as error:
            status = error.response.status_code if error.response is not None else None
            if status == 403:
                raise PermissionError(f"Access denied: {cloud_path.path}") from error
            raise ConnectionError(f"Upload failed: {error}") from error
        except requests.exceptions.RequestException as error:
            raise ConnectionError(f"Upload failed: {error}") from error




    def download(self, cloud_path: CloudPath, destination: Path) -> Path:
        """
        Download a file via HTTPS GET.

        Parameters
        ----------
        cloud_path : CloudPath
            The HTTPS source URL.
        destination : Path
            The local destination path.

        Returns
        -------
        Path
            The path to the downloaded file.

        Raises
        ------
        FileNotFoundError
            If the remote file does not exist (404).
        ConnectionError
            If the download fails.
        PermissionError
            If access is denied (403).
        """
        try:
            response = requests.get(cloud_path.full_url, stream=True, timeout=300)
            response.raise_for_status()

            destination.parent.mkdir(parents=True, exist_ok=True)
            with open(destination, "wb") as file_handle:
                for chunk in response.iter_content(chunk_size=8192):
                    if chunk:
                        file_handle.write(chunk)
            return destination

        except requests.exceptions.HTTPError as error:
            status = error.response.status_code if error.response is not None else None
            if status == 403:
                raise PermissionError(f"Access denied: {cloud_path.path}") from error
            if status == 404:
                raise FileNotFoundError(f"Not found: {cloud_path.path}") from error
            raise ConnectionError(f"Download failed: {error}") from error
        except requests.exceptions.RequestException as error:
            raise ConnectionError(f"Download failed: {error}") from error




    def file_exists(self, cloud_path: CloudPath) -> bool:
        """
        Check if an HTTPS resource exists using HEAD request.

        Parameters
        ----------
        cloud_path : CloudPath
            The HTTPS URL to check.

        Returns
        -------
        bool
            True if the resource exists (HTTP 200), False otherwise.
        """
        try:
            response = requests.head(cloud_path.full_url, timeout=10)
            return bool(response.status_code == 200)
        except requests.exceptions.RequestException:
            return False