Source code for shapelib.core.ports.file_writer_port

"""Port (Protocol) for file writing strategies."""

from __future__ import annotations

from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable

if TYPE_CHECKING:
    import geopandas as gpd

    from shapelib.core.cloud.cloud_path import CloudPath


[docs] @runtime_checkable class FileWriterStrategy(Protocol): """ Protocol defining the interface for file writing strategies. This protocol follows the Strategy pattern, allowing different implementations for saving geospatial data to various destinations (local filesystem, cloud storage, etc.). Implementations must provide: - `supports()`: Check if this strategy handles the given path - `write()`: Perform the actual write operation Examples -------- Implementing a custom writer: >>> class MyCloudWriter: ... def supports(self, cloud_path: CloudPath) -> bool: ... return cloud_path.scheme == "mycloud" ... ... def write( ... self, ... gdf: gpd.GeoDataFrame, ... cloud_path: CloudPath, ... **kwargs: Any, ... ) -> None: ... # Upload to MyCloud ... pass Notes ----- This protocol enables Clean Architecture by allowing the Domain (BaseShapes) to depend on an abstraction rather than concrete infrastructure implementations. """
[docs] def supports(self, cloud_path: CloudPath) -> bool: """ Check if this strategy supports the given path. Parameters ---------- cloud_path : CloudPath The parsed destination path. Returns ------- bool True if this strategy can handle the path. """ ...
[docs] def write( self, gdf: gpd.GeoDataFrame, cloud_path: CloudPath, driver: str | None = None, schema: Any | None = None, index: Any | None = None, **kwargs: Any, ) -> None: """ Write the GeoDataFrame to the destination. Parameters ---------- gdf : gpd.GeoDataFrame The GeoDataFrame to save. cloud_path : CloudPath The parsed destination path. driver : str | None The file driver (e.g., 'FlatGeobuf'). schema : Any | None The schema definition. index : Any | None Whether to write the index. **kwargs : Any Additional arguments. Raises ------ ConnectionError If the write operation fails due to network issues. PermissionError If access is denied. """ ...