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

"""Remote handler for GeoJSON 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(".geojson")
@register_cloud_handler(".json")
class RemoteGeojsonHandler(RemoteGeoFileHandler):
    """
    Handler for remote GeoJSON files using GDAL streaming.

    GeoJSON files can be read via streaming, though the entire JSON
    structure typically needs to be parsed. For very large files,
    consider using FlatGeobuf instead.

    Attributes
    ----------
    driver : str
        GDAL driver name: "GeoJSON".
    extension : str
        File extension: ".geojson" or ".json".

    Examples
    --------
    >>> from shapelib import read_file
    >>> polygons = read_file("https://example.com/data.geojson")

    Big-O
    -----
    O(n) - full file read required for JSON parsing.
    """

    driver: str = "GeoJSON"
    extension: str = ".geojson"



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

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

        Returns
        -------
        gpd.GeoDataFrame
            The loaded geospatial data.
        """
        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