Source code for shapelib.core.cloud.handlers.remote_flatgeobuf_handler

"""Remote handler for FlatGeobuf files using streaming."""

from typing import Any

import geopandas as gpd

from shapelib.core.cloud.handlers.remote_handler import RemoteGeoFileHandler
from shapelib.factories.cloud_registry import register_cloud_handler




@register_cloud_handler(".fgb")
class RemoteFlatGeobufHandler(RemoteGeoFileHandler):
    """
    Handler for remote FlatGeobuf files using GDAL streaming.

    FlatGeobuf is optimized for streaming with HTTP range request support,
    making it ideal for cloud storage. This handler uses GDAL's /vsicurl/
    to read data directly without downloading the entire file.

    Attributes
    ----------
    driver : str
        GDAL driver name: "FlatGeobuf".
    extension : str
        File extension: ".fgb".

    Examples
    --------
    >>> from shapelib import read_file
    >>> lines = read_file("https://bucket.s3.amazonaws.com/file.fgb")
    >>> print(type(lines))
    <class 'shapelib.core.geometries.lines.Lines'>

    Big-O
    -----
    O(n) with efficient range requests - no full download required.
    """

    driver: str = "FlatGeobuf"
    extension: str = ".fgb"



    def load_data(self, **kwargs: Any) -> gpd.GeoDataFrame:
        """
        Load FlatGeobuf data via GDAL streaming.

        Parameters
        ----------
        **kwargs : Any
            Additional arguments passed to geopandas.read_file().

        Returns
        -------
        gpd.GeoDataFrame
            The loaded geospatial data.

        Raises
        ------
        ConnectionError
            If the remote file cannot be accessed.
        FileNotFoundError
            If the remote file does not exist.
        """
        try:
            return gpd.read_file(self.cloud_path.vsicurl_path, **kwargs)
        except Exception as error:
            self._handle_read_error(error)


    def _handle_read_error(self, error: Exception) -> None:
        """Convert GDAL errors to appropriate exceptions."""
        message = str(error).lower()
        if "404" in message or "not found" in message:
            raise FileNotFoundError(f"Remote file not found: {self.cloud_path.path}") from error
        if "403" in message or "denied" in message:
            raise PermissionError(f"Access denied: {self.cloud_path.path}") from error
        raise ConnectionError(f"Failed to read remote file: {error}") from error