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.
"""
...