Source code for shapelib.core.cloud.handlers.remote_shapefile_handler
"""Remote handler for Shapefile files using download."""
from __future__ import annotations
import contextlib
import tempfile
from pathlib import Path
from typing import TYPE_CHECKING, Any
import geopandas as gpd
from shapelib.core.cloud.cloud_path import CloudPath
from shapelib.core.cloud.handlers.remote_handler import RemoteGeoFileHandler
from shapelib.factories.cloud_registry import register_cloud_handler
if TYPE_CHECKING:
from shapelib.factories.geo_cloud_factory import GeoCloudFactory
[docs]
@register_cloud_handler(".shp")
class RemoteShapefileHandler(RemoteGeoFileHandler):
"""
Handler for remote Shapefile files using download.
Shapefiles require multiple component files (.shp, .shx, .dbf, .prj, .cpg).
This handler downloads all sidecar files to a temporary directory before
reading.
Note: For cloud storage, FlatGeobuf or GeoPackage are recommended
as they are single-file formats with streaming support.
Attributes
----------
driver : str
GDAL driver name: "ESRI Shapefile".
extension : str
File extension: ".shp".
sidecar_extensions : tuple[str, ...]
Required and optional sidecar file extensions.
Examples
--------
>>> from shapelib import read_file
>>> polygons = read_file("https://example.com/data.shp")
Big-O
-----
O(n) with full download of all component files required.
"""
driver: str = "ESRI Shapefile"
extension: str = ".shp"
sidecar_extensions: tuple[str, ...] = (".shx", ".dbf", ".prj", ".cpg")
[docs]
def load_data(self, **kwargs: Any) -> gpd.GeoDataFrame:
"""
Load Shapefile data by downloading all component files.
Downloads .shp and all sidecar files (.shx, .dbf, .prj, .cpg)
to a temporary directory before reading.
For s3:// URLs, sidecar files are downloaded automatically.
For HTTPS presigned URLs, each sidecar needs its own presigned URL,
so this may fail. Use s3:// protocol or FlatGeobuf format instead.
Parameters
----------
**kwargs : Any
Additional arguments passed to geopandas.read_file().
Returns
-------
gpd.GeoDataFrame
The loaded geospatial data.
"""
from shapelib.factories.geo_cloud_factory import GeoCloudFactory
with tempfile.TemporaryDirectory() as temp_dir:
shp_filename = Path(self.cloud_path.path).name
shp_stem = Path(shp_filename).stem
temp_shp = Path(temp_dir) / shp_filename
# Download main .shp file
GeoCloudFactory.download_file(self.cloud_path, temp_shp)
# Download sidecar files
self._download_sidecar_files(temp_dir, shp_stem, GeoCloudFactory)
return gpd.read_file(temp_shp, **kwargs)
def _download_sidecar_files(self, temp_dir: str, shp_stem: str, factory: type[GeoCloudFactory]) -> None:
"""Download shapefile sidecar files (.shx, .dbf, .prj, .cpg)."""
for ext in self.sidecar_extensions:
sidecar_url = self._build_sidecar_url(ext)
sidecar_cloud_path = CloudPath.from_string(sidecar_url)
# Check if sidecar file exists before attempting download
if not factory.file_exists(sidecar_cloud_path):
continue
sidecar_path = Path(temp_dir) / f"{shp_stem}{ext}"
with contextlib.suppress(FileNotFoundError, ConnectionError, PermissionError):
factory.download_file(sidecar_cloud_path, sidecar_path)
def _build_sidecar_url(self, extension: str) -> str:
"""Build URL for a sidecar file by replacing .shp extension."""
if self.cloud_path.is_s3_protocol:
# s3://bucket/path/file.shp -> s3://bucket/path/file.ext
return self.cloud_path.raw_path.rsplit(".shp", 1)[0] + extension
# For HTTPS, replace extension in path, preserve query string
base_path = self.cloud_path.path.rsplit(".shp", 1)[0] + extension
if self.cloud_path.query:
return f"{self.cloud_path.scheme}://{self.cloud_path.host}{base_path}?{self.cloud_path.query}"
return f"{self.cloud_path.scheme}://{self.cloud_path.host}{base_path}"