Source code for shapelib.factories.cloud_registry

"""Decorators for registering cloud components (handlers, providers, writers)."""

from collections.abc import Callable
from typing import Any




def register_cloud_handler(file_extension: str) -> Callable[[type[Any]], type[Any]]:
    """
    Decorator to register a remote file handler with GeoCloudFactory.

    This decorator associates a file extension with a remote handler class,
    enabling GeoCloudFactory to automatically select the correct handler
    for remote files based on their extension.

    Parameters
    ----------
    file_extension : str
        The file extension this handler supports (e.g., ".fgb", ".shp").

    Returns
    -------
    Callable[[type[Any]], type[Any]]
        A decorator that registers the handler class.

    Examples
    --------
    >>> @register_cloud_handler(".fgb")
    ... class RemoteFlatGeobufHandler(RemoteGeoFileHandler):
    ...     driver = "FlatGeobuf"
    ...     extension = ".fgb"
    ...
    ...     def load_data(self, **kwargs):
    ...         return gpd.read_file(self.cloud_path.vsicurl_path, **kwargs)
    """

    def decorator(handler_class: type[Any]) -> type[Any]:
        """Register the handler class with GeoCloudFactory."""
        # Import here to avoid circular import
        from shapelib.factories.geo_cloud_factory import GeoCloudFactory

        GeoCloudFactory.register_handler(file_extension, handler_class)
        return handler_class

    return decorator





def register_cloud_provider(cls: type[Any]) -> type[Any]:
    """
    Decorator to register a cloud provider with GeoCloudFactory.

    This decorator registers a cloud provider class (S3, HTTPS, GCS, etc.)
    enabling GeoCloudFactory to automatically select the correct provider
    for cloud operations based on the URL scheme.

    Parameters
    ----------
    cls : type[Any]
        The provider class to register.

    Returns
    -------
    type[Any]
        The registered provider class.

    Examples
    --------
    >>> @register_cloud_provider
    ... class S3Provider:
    ...     def supports(self, cloud_path: CloudPath) -> bool:
    ...         return cloud_path.is_s3_protocol
    ...
    ...     def upload(self, source: Path, cloud_path: CloudPath) -> None:
    ...         # S3 upload logic
    ...         ...
    """
    # Import here to avoid circular import
    from shapelib.factories.geo_cloud_factory import GeoCloudFactory

    GeoCloudFactory.register_provider(cls())
    return cls





def register_cloud_writer(cls: type[Any]) -> type[Any]:
    """
    Decorator to register a cloud writer with BaseShapes.

    This decorator registers a writer class for cloud file operations,
    enabling BaseShapes.save() to automatically select the correct writer
    based on the file format and destination.

    Parameters
    ----------
    cls : type[Any]
        The writer class to register.

    Returns
    -------
    type[Any]
        The registered writer class.

    Examples
    --------
    >>> @register_cloud_writer
    ... class FlatGeobufCloudWriter(BaseCloudWriter):
    ...     def supports(self, cloud_path: CloudPath) -> bool:
    ...         return cloud_path.is_remote and cloud_path.extension == ".fgb"
    ...
    ...     def _build_write_kwargs(self, ...) -> dict:
    ...         return {"engine": "pyogrio", ...}
    """
    # Import here to avoid circular import
    from shapelib.core.geometries.base import BaseShapes

    BaseShapes.register_writer(cls())
    return cls