Source code for shapelib.core.cloud.writers.base_cloud_writer

"""Base class for cloud file writers."""

from __future__ import annotations

import tempfile
from abc import ABC, abstractmethod
from pathlib import Path
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    import geopandas as gpd

from shapelib.core.cloud.cloud_path import CloudPath




class BaseCloudWriter(ABC):
    """
    Abstract base class for cloud file writers.

    Each concrete writer handles a specific file format with
    format-specific optimizations for cloud storage.

    Subclasses must implement:
    - `supports()`: Check if this writer handles the given path
    - `_build_write_kwargs()`: Build format-specific kwargs

    Examples
    --------
    >>> class MyFormatWriter(BaseCloudWriter):
    ...     def supports(self, cloud_path: CloudPath) -> bool:
    ...         return cloud_path.extension == ".myformat"
    ...
    ...     def _build_write_kwargs(self, ...) -> dict[str, Any]:
    ...         return {"driver": "MyFormat", **kwargs}
    """



    @abstractmethod
    def supports(self, cloud_path: CloudPath) -> bool:
        """
        Check if this writer supports the given path.

        Parameters
        ----------
        cloud_path : CloudPath
            The parsed destination path.

        Returns
        -------
        bool
            True if this writer handles the format.
        """
        ...




    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 GeoDataFrame to cloud storage.

        Parameters
        ----------
        gdf : gpd.GeoDataFrame
            The GeoDataFrame to save.
        cloud_path : CloudPath
            The remote destination path.
        driver : str | None
            The file driver.
        schema : Any | None
            The schema definition.
        index : Any | None
            Whether to write the index.
        **kwargs : Any
            Additional arguments passed to to_file().

        Raises
        ------
        ConnectionError
            If the upload fails.
        PermissionError
            If access is denied.
        """
        from shapelib.factories.geo_cloud_factory import GeoCloudFactory

        with tempfile.TemporaryDirectory() as temp_dir:
            filename = Path(cloud_path.path).name
            temp_file = Path(temp_dir) / filename

            write_kwargs = self._build_write_kwargs(temp_dir, driver, schema, index, **kwargs)
            gdf.to_file(filename=temp_file, **write_kwargs)
            GeoCloudFactory.upload_file(temp_file, cloud_path.full_url)


    @abstractmethod
    def _build_write_kwargs(
        self,
        temp_dir: str,
        driver: str | None,
        schema: Any | None,
        index: Any | None,
        **kwargs: Any,
    ) -> dict[str, Any]:
        """
        Build kwargs for to_file() with format-specific settings.

        Parameters
        ----------
        temp_dir : str
            Temporary directory path (for formats needing temp space).
        driver : str | None
            The file driver.
        schema : Any | None
            The schema definition.
        index : Any | None
            Whether to write the index.
        **kwargs : Any
            Additional user-provided arguments.

        Returns
        -------
        dict[str, Any]
            Complete kwargs for gdf.to_file().
        """
        ...