Implement Koordinates API integration with export validation, layer d…

.env.template

@@ -0,0 +1,4 @@
+WORKSPACEFOLDER="${YourWorkSpaceHere}"
+PYTHONPATH="${WORKSPACEFOLDER}/src"
+KOOPCACHE_DIR="${WORKSPACEFOLDER}/.koopcache"
+KOORDINATES_API_KEY="**"

.vscode/settings.json

@@ -2,11 +2,11 @@
     "python.testing.pytestEnabled": true,
     "python.testing.unittestEnabled": false,
     "python.testing.pytestArgs": [
-        "tests/"
+        "tests"
     ],
     "python.defaultInterpreterPath": "./.venv/Scripts/python.exe",
     "python.terminal.activateEnvironment": true,
     "pythonTestExplorer.testFramework": "pytest",
     "pythonTestExplorer.testplanEnabled": false,
     "testExplorer.useNativeTesting": true
-}
+}

pyproject.toml

@@ -12,7 +12,7 @@ requires = [
 ]
 
 [project]
-name = "whitelabel"
+name = "koop"
 description = "A minimal Python package template with a simple Hello World example."
 readme = "README.md"
 license = "MIT"
@@ -33,7 +33,11 @@ classifiers = [
   "Programming Language :: Python :: 3.13",
 ]
 dynamic = [ "urls", "version" ]
-dependencies = [  ]
+dependencies = [
+  "defusedxml>=0.7.1",
+  "pydantic>=2.0.0",
+  "requests>=2.0.0",
+]
 
 [dependency-groups]
 dev = [
@@ -74,7 +78,7 @@ source = "vcs"
 "Source Archive" = "https://github.com/tonkintaylor/YOUR_REPOSITORY/archive/{commit_hash}.zip"
 
 [tool.hatch.build.hooks.vcs]
-version-file = "src/whitelabel/_version.py"
+version-file = "src/koop/_version.py"
 
 [tool.ruff]
 line-length = 88
@@ -229,5 +233,5 @@ output = "test-reports/coverage.xml"
 
 [tool.uv]
 default-groups = [ "dev", "test", "doc" ]
-required-version = "==0.8.3"              # sync with .pre-commit-config.yaml and release.yml
+required-version = ">=0.7.13"              # sync with .pre-commit-config.yaml and release.yml
 link-mode = "symlink"

src/koop/__init__.py

@@ -0,0 +1,32 @@
+from pathlib import Path
+
+from koop.backend.conn import KoordinatesConnection
+from koop.get_latest import get_latest_layer
+
+__all__ = ["get_layer_from_id"]
+
+
+def get_layer_from_id(
+    layer_id: int, api_key: str, domain: str = "ttgroup.koordinates.com"
+) -> Path:
+    """Helper function to get the latest layer from Koordinates.
+
+    Args:
+        layer_id (int): The layer ID.
+        api_key (str): Your API key.
+        domain (str, optional): The domain of the API.
+
+    Returns:
+        Path: The path to the latest layer.
+
+    """
+    conn = KoordinatesConnection(
+        api_key,
+        domain,
+    )
+
+    try:
+        return get_latest_layer(conn=conn, layer_id=layer_id)
+    finally:
+        if conn:
+            conn.close()

src/koop/backend/api/download_layer.py

@@ -0,0 +1,136 @@
+from pathlib import Path
+from time import sleep
+from zipfile import ZipFile
+
+from requests import Session
+
+from koop.backend.api.exports.create_export import create_export
+from koop.backend.api.exports.get_exports import get_export_from_id
+from koop.backend.api.exports.validate_export import validate_export
+from koop.backend.api.layers_and_tables.get_details import get_layer_details
+
+
+def download_layer(
+    *,
+    session: Session,
+    domain: str,
+    api_version: str,
+    layer_id: int,
+    output_dir: Path,
+) -> Path:
+    """Download a layer from the Koordinates server.
+
+    Args:
+        session: The session to use to make the request.
+        domain: The domain to use for the request.
+        api_version: The version of the API to use for the request.
+        layer_id: The ID of the layer to download.
+        output_dir: The directory to save the downloaded layer to.
+
+    Returns:
+        The path to the directory whether the layer is saved.
+
+    """
+    layer_details = get_layer_details(session, domain, api_version, layer_id)
+
+    # validate
+    validation_response = validate_export(
+        session,
+        domain,
+        api_version,
+        layer_id,
+        export_format=layer_details.kind.value,
+    )
+
+    if not validation_response.is_valid:
+        raise ValueError(validation_response.invalid_reasons)
+
+    export = create_export(
+        session,
+        domain,
+        api_version,
+        layer_id,
+        export_format=layer_details.kind.value,
+    )
+
+    # wait for export to complete
+    export_url = wait_for_export(session, domain, api_version, export.id)
+
+    # download
+    filename = download_export(session, export_url, output_dir)
+
+    # unzip
+    unzipped_dir = unzip_export(output_dir / filename)
+
+    return unzipped_dir
+
+
+def wait_for_export(
+    session: Session, domain: str, api_version: str, export_id: int
+) -> str:
+    """Wait for the export to complete.
+
+    Args:
+        session: The session to use to make the request.
+        domain: The domain to use for the request.
+        api_version: The version of the API to use for the request.
+        export_id: The ID of the export to wait for.
+
+    """
+    state = "processing"
+    while state != "complete":
+        sleep(1)
+
+        export = get_export_from_id(session, domain, api_version, export_id)
+        state = export.state
+
+        if state not in ["processing", "complete"]:
+            msg = f"Export failed with state: {state}"
+            raise ValueError(msg)
+
+    return export.download_url
+
+
+def download_export(session: Session, export_url: str, output_dir: Path) -> Path:
+    """Download the export from the Koordinates server.
+
+    Args:
+        session: The session to use to make the request.
+        domain: The domain to use for the request.
+        api_version: The version of the API to use for the request.
+        export_url: The URL of the export to download.
+        output_dir: The directory to save the downloaded export to.
+
+    """
+    with session.get(export_url, stream=True) as response:
+        response.raise_for_status()
+
+        if "Content-Disposition" in response.headers:
+            filename = response.headers["Content-Disposition"].split(
+                "filename*=UTF-8''"
+            )[1]
+        else:
+            filename = "export.zip"
+
+        with (output_dir / filename).open("wb") as file:
+            for chunk in response.iter_content(chunk_size=1024):
+                file.write(chunk)
+
+    return filename
+
+
+def unzip_export(zip_path: Path) -> Path:
+    """Unzip the export.
+
+    Args:
+        zip_path: The path to the export to unzip.
+
+    """
+    output_dir = zip_path.parent / zip_path.stem
+
+    with ZipFile(zip_path, "r") as zip_ref:
+        zip_ref.extractall(output_dir)
+
+    zip_path.unlink()
+
+    return output_dir

src/koop/backend/api/exports/create_export.py

@@ -0,0 +1,64 @@
+from pydantic import BaseModel
+from requests import Session
+
+from koop.backend.api.exports.post_export import post_export
+
+
+def create_export(  # noqa: PLR0913
+    session: Session,
+    domain: str,
+    api_version: str,
+    layer_id: int,
+    export_format: str,
+    file_type: str | None = None,
+    tiles: list[str] | None = None,
+) -> dict:
+    """Create and start a new export.
+
+    https://apidocs.koordinates.com/#tag/Exports/operation/postExport
+
+    Args:
+        session: The session to use to make the request.
+        domain: The domain to use to make the request.
+        api_version: The version of the API to use.
+        layer_id: The ID of the layer to export.
+        export_format: The format to export the layer to.
+        file_type: The file type to export the layer to. Defaults to None.
+        tiles: The tiles to export. Defaults to None.
+
+    Returns:
+        dict: The JSON response.
+    """
+    url = f"https://{domain}/services/api/v{api_version}/exports/"
+
+    response = post_export(
+        session, url, domain, api_version, layer_id, export_format, file_type, tiles
+    )
+
+    return _ResponseSchema(**response.json())
+
+
+class _ResponseSchema(BaseModel):
+    """The response schema."""
+
+    id: int
+    name: str
+    created_at: str | None
+    created_via: str
+    state: str
+    url: str
+    download_url: str | None
+    user: dict
+    delivery: dict
+    items: list[dict]
+    crs: dict
+    extent: str | None
+    formats: dict
+    options: dict | None
+    size_estimate_unzipped: int
+    size_complete_zipped: int | None
+    size_complete_unzipped: int | None
+    is_cropped: bool
+    invoice: str | None
+    _from: dict
+    progress: float

src/koop/backend/api/exports/enums.py

@@ -0,0 +1,21 @@
+from enum import Enum
+from typing import ClassVar
+
+
+class GridExportFormats(Enum):
+    """Grid Export Formats enumeration."""
+
+    geotiff = "image/tiff;subtype=geotiff"
+
+
+class VectorExportFormats(Enum):
+    """Vector Export Formats enumeration."""
+
+    geopackage = "application/x-ogc-gpkg"
+
+
+class ExportFormats(dict, Enum):
+    """Represents the export formats for raster and vector data."""
+
+    grid: ClassVar = {"grid": GridExportFormats.geotiff.value}
+    vector: ClassVar = {"vector": VectorExportFormats.geopackage.value}

src/koop/backend/api/exports/get_exports.py

@@ -0,0 +1,56 @@
+from datetime import datetime
+
+from pydantic import BaseModel, TypeAdapter
+from requests import Session
+
+
+class _ResponseSchema(BaseModel):
+    """The export schema."""
+
+    id: int
+    name: str
+    created_at: datetime
+    state: str
+    url: str
+    download_url: str | None
+
+
+def get_exports(session: Session, domain: str, api_version: str) -> list:
+    """Returns a list of exports you've created.
+
+    https://apidocs.koordinates.com/#tag/Exports/operation/getExports
+
+    Args:
+        session (requests.Session): The session to use to make the request.
+        domain (str): The domain to use to make the request.
+        api_version (str): The version of the API to use.
+
+    Returns:
+        list: The JSON response.
+    """
+    url = f"https://{domain}/services/api/v{api_version}/exports/"
+
+    with session.get(url) as response:
+        response.raise_for_status()
+        return TypeAdapter(list[_ResponseSchema]).validate_python(response.json())
+
+
+def get_export_from_id(
+    session: Session, domain: str, api_version: str, export_id: int
+) -> _ResponseSchema:
+    """Returns a list of exports you've created.
+
+    Args:
+        session (requests.Session): The session to use to make the request.
+        domain (str): The domain to use to make the request.
+        api_version (str): The version of the API to use.
+        export_id (int): The ID of the export to get.
+
+    Returns:
+        list: The JSON response.
+    """
+    url = f"https://{domain}/services/api/v{api_version}/exports/{export_id}/"
+
+    with session.get(url) as response:
+        response.raise_for_status()
+        return _ResponseSchema(**response.json())