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

"""Protocol defining the interface for cloud storage providers."""

from __future__ import annotations

from pathlib import Path
from typing import TYPE_CHECKING, Protocol, runtime_checkable

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


[docs] @runtime_checkable class CloudProvider(Protocol): """ Protocol defining the interface for cloud storage providers. Each cloud provider (S3, HTTPS, GCS, Azure) implements this interface to handle file operations for their specific storage system. This follows the Strategy pattern, allowing GeoCloudFactory to delegate operations to the appropriate provider without conditional logic. Examples -------- >>> class MyCloudProvider: ... def supports(self, cloud_path: CloudPath) -> bool: ... return cloud_path.scheme == "mycloud" ... ... def upload(self, source: Path, cloud_path: CloudPath) -> None: ... # Upload implementation ... ... Big-O ----- All operations are O(n) where n is the file size. """
[docs] def supports(self, cloud_path: CloudPath) -> bool: """ Check if this provider handles the given cloud path. Parameters ---------- cloud_path : CloudPath The parsed cloud path to check. Returns ------- bool True if this provider can handle the path. """ ...
[docs] def upload(self, source: Path, cloud_path: CloudPath) -> None: """ Upload a local file to cloud storage. Parameters ---------- source : Path The local file to upload. cloud_path : CloudPath The remote destination path. Raises ------ FileNotFoundError If the source file does not exist. ConnectionError If the upload fails. PermissionError If access is denied. """ ...
[docs] def download(self, cloud_path: CloudPath, destination: Path) -> Path: """ Download a remote file to local storage. Parameters ---------- cloud_path : CloudPath The remote source path. destination : Path The local destination path. Returns ------- Path The path to the downloaded file. Raises ------ FileNotFoundError If the remote file does not exist. ConnectionError If the download fails. PermissionError If access is denied. """ ...
[docs] def file_exists(self, cloud_path: CloudPath) -> bool: """ Check if a remote file exists without downloading. Parameters ---------- cloud_path : CloudPath The remote file path to check. Returns ------- bool True if the file exists, False otherwise. """ ...