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.
"""
...