Extend cloud support#

shapelib’s cloud layer is registry-based: providers, writers and remote handlers register themselves with decorators, so adding a new cloud (e.g. Google Cloud Storage or Azure Blob) means writing one class — no core changes.

Current state:

Provider	Protocol	Status
AWS S3	`s3://`	supported
HTTPS (presigned URLs)	`https://`	supported
Google Cloud Storage	`gs://`	not yet
Azure Blob Storage	`az://`	not yet

Step 1 — write the provider#

Create src/shapelib/core/cloud/providers/gcs_provider.py implementing supports / upload / download / file_exists, decorated with @register_cloud_provider:

from pathlib import Path

from shapelib.core.cloud.cloud_path import CloudPath
from shapelib.factories.cloud_registry import register_cloud_provider


@register_cloud_provider  # auto-registers on import
class GcsProvider:
    """Google Cloud Storage provider."""

    def supports(self, cloud_path: CloudPath) -> bool:
        return cloud_path.scheme == "gs"

    def upload(self, source: Path, cloud_path: CloudPath) -> None:
        from google.cloud import storage

        client = storage.Client()
        bucket = client.bucket(cloud_path.host)
        bucket.blob(cloud_path.path.lstrip("/")).upload_from_filename(str(source))

    def download(self, cloud_path: CloudPath, destination: Path) -> Path:
        from google.cloud import storage

        client = storage.Client()
        bucket = client.bucket(cloud_path.host)
        destination.parent.mkdir(parents=True, exist_ok=True)
        bucket.blob(cloud_path.path.lstrip("/")).download_to_filename(str(destination))
        return destination

    def file_exists(self, cloud_path: CloudPath) -> bool:
        from google.cloud import storage

        client = storage.Client()
        bucket = client.bucket(cloud_path.host)
        return bucket.blob(cloud_path.path.lstrip("/")).exists()

Step 2 — trigger the registration#

The decorator runs on import, so import the module at the end of src/shapelib/factories/geo_cloud_factory.py:

import shapelib.core.cloud.providers.gcs_provider  # noqa: F401, E402

Step 3 (optional) — streaming support#

For formats that stream (.fgb, .geojson, .gpkg), teach CloudPath.vsicurl_path (src/shapelib/core/cloud/cloud_path.py) how to build the GDAL virtual path for the new scheme — e.g. /vsigs/<bucket>/<key> for GCS.

Step 4 — dependency and tests#

Add the SDK to the project dependencies (google-cloud-storage, azure-storage-blob, …) and cover the provider in tests/test_cloud_support.py, skipping when credentials are absent:

@pytest.mark.skipif(not has_gcs_credentials(), reason="GCS credentials not available")
class TestGCSIntegration:
    def test_read_flatgeobuf_gcs(self):
        result = shapelib.read_file("gs://my-bucket/data.fgb")

Where things live#

File	Purpose
`src/shapelib/factories/cloud_registry.py`	The decorators: `@register_cloud_provider`, `@register_cloud_writer`, `@register_cloud_handler`
`src/shapelib/core/cloud/providers/`	Cloud provider implementations
`src/shapelib/core/cloud/writers/`	Format-specific cloud writers
`src/shapelib/core/cloud/cloud_path.py`	URL parsing and protocol detection