CLI Architecture & Design Patterns for Python GIS

# cli/main.py — interface layer only; no rasterio imports here
from pathlib import Path
import typer
from myapp.orchestration import run_clip_job
from myapp.models import ClipConfig

app = typer.Typer(help="GIS batch processor")

@app.command()
def clip(
    src: Path = typer.Argument(..., help="Input GeoTIFF (EPSG:4326 expected)"),
    aoi: Path = typer.Argument(..., help="GeoJSON area-of-interest polygon"),
    dst: Path = typer.Option(Path("output.tif"), help="Destination path"),
    crs: str = typer.Option("EPSG:4326", help="Target CRS as EPSG string"),
) -> None:
    # Validate at the boundary before touching the filesystem
    if not src.exists():
        raise typer.BadParameter(f"Source file not found: {src}", param_hint="src")
    if not aoi.exists():
        raise typer.BadParameter(f"AOI file not found: {aoi}", param_hint="aoi")

    config = ClipConfig(src=src, aoi=aoi, dst=dst, crs=crs)
    # Delegate to orchestration; interface layer is now done
    run_clip_job(config)

# orchestration/clip.py — sequences steps, handles retries, owns exit codes
import os
import tempfile
from pathlib import Path
from myapp.engine import clip_raster_to_aoi
from myapp.models import ClipConfig
from myapp.logging import get_logger

log = get_logger(__name__)

def run_clip_job(config: ClipConfig) -> None:
    log.info("clip_start", src=str(config.src), crs=config.crs)
    # Atomic write: produce to temp path, rename on success
    with tempfile.NamedTemporaryFile(
        suffix=".tif", dir=config.dst.parent, delete=False
    ) as tmp:
        tmp_path = Path(tmp.name)

    try:
        clip_raster_to_aoi(config.src, config.aoi, tmp_path, config.crs)
        os.replace(tmp_path, config.dst)      # atomic rename prevents partial reads
        log.info("clip_done", dst=str(config.dst))
    except Exception:
        tmp_path.unlink(missing_ok=True)      # clean up on failure
        raise

# engine/raster.py — pure domain logic; no CLI dependencies
import rasterio
import rasterio.mask
import geopandas as gpd
from pathlib import Path
from pyproj import CRS

def clip_raster_to_aoi(
    src: Path, aoi: Path, dst: Path, target_crs: str = "EPSG:4326"
) -> None:
    aoi_gdf = gpd.read_file(aoi).to_crs(target_crs)  # reproject AOI to raster CRS
    shapes = [geom.__geo_interface__ for geom in aoi_gdf.geometry]

    with rasterio.open(src) as raster:
        # Verify CRS compatibility before starting the expensive operation
        src_crs = CRS.from_user_input(raster.crs)
        target = CRS.from_user_input(target_crs)
        if not src_crs.equals(target):
            raise ValueError(
                f"CRS mismatch: source is {src_crs.to_epsg()}, "
                f"expected {target.to_epsg()}"
            )
        out_image, out_transform = rasterio.mask.mask(raster, shapes, crop=True)
        out_meta = raster.meta.copy()
        out_meta.update({
            "height": out_image.shape[1],
            "width":  out_image.shape[2],
            "transform": out_transform,
        })

    with rasterio.open(dst, "w", **out_meta) as dest:
        dest.write(out_image)

# engine/windowed.py — generator-based raster pipeline
from typing import Generator, Iterator
from pathlib import Path
import numpy as np
import rasterio
from rasterio.windows import Window

def iter_raster_blocks(
    src_path: Path,
) -> Generator[tuple[Window, np.ndarray], None, None]:
    """Yield (window, data) pairs aligned to the file's internal block grid."""
    with rasterio.open(src_path) as src:
        # block_shapes returns a list of (height, width) per band; use band 1
        block_h, block_w = src.block_shapes[0]
        for row_off in range(0, src.height, block_h):
            for col_off in range(0, src.width, block_w):
                # Clamp window to raster extent at edges
                win = Window(
                    col_off=col_off,
                    row_off=row_off,
                    width=min(block_w, src.width - col_off),
                    height=min(block_h, src.height - row_off),
                )
                yield win, src.read(window=win)   # data stays on disk until here

def apply_ndvi_pipeline(src_path: Path, dst_path: Path) -> None:
    """Compute NDVI band-by-band without loading the full raster."""
    with rasterio.open(src_path) as src:
        profile = src.profile.copy()
        profile.update(count=1, dtype="float32")

    import tempfile, os
    with tempfile.NamedTemporaryFile(
        suffix=".tif", dir=dst_path.parent, delete=False
    ) as tmp:
        tmp_path = Path(tmp.name)

    try:
        with rasterio.open(tmp_path, "w", **profile) as dst:
            for win, data in iter_raster_blocks(src_path):
                red  = data[2].astype("float32")   # band index depends on sensor
                nir  = data[3].astype("float32")
                ndvi = (nir - red) / (nir + red + 1e-10)   # avoid div-by-zero
                dst.write(ndvi[np.newaxis, ...], window=win)
        os.replace(tmp_path, dst_path)
    except Exception:
        Path(tmp_path).unlink(missing_ok=True)
        raise

# cli/app.py — hierarchical command tree with lazy imports
import typer

app = typer.Typer(help="GIS batch processor")
raster_app = typer.Typer(help="Raster operations")
vector_app = typer.Typer(help="Vector operations")

app.add_typer(raster_app, name="raster")
app.add_typer(vector_app, name="vector")

@raster_app.command("clip")
def raster_clip(src: str, aoi: str) -> None:
    # Heavy imports deferred until this subcommand is actually invoked
    from myapp.engine.raster import clip_raster_to_aoi  # noqa: PLC0415
    ...

@vector_app.command("buffer")
def vector_buffer(src: str, distance: float) -> None:
    from myapp.engine.vector import buffer_features     # noqa: PLC0415
    ...

# config/loader.py — precedence chain with schema validation
from pathlib import Path
from typing import Any
import yaml
from pydantic import BaseModel, Field

class AppConfig(BaseModel):
    gdal_cachemax:   int  = Field(512,           description="GDAL block cache in MB")
    output_crs:      str  = Field("EPSG:4326",   description="Default output CRS")
    chunk_size:      int  = Field(1024,           description="Raster window size in pixels")
    workers:         int  = Field(4,              description="Parallel worker count")
    log_format:      str  = Field("json",         description="'json' or 'text'")

def load_config(config_path: Path | None = None) -> AppConfig:
    """Merge defaults, YAML file, and env vars in precedence order."""
    import os

    # 1. Start from model defaults
    data: dict[str, Any] = {}

    # 2. Overlay YAML file if provided or found at default location
    search = config_path or Path.home() / ".config" / "geotools" / "config.yaml"
    if search.exists():
        with search.open() as f:
            data.update(yaml.safe_load(f) or {})

    # 3. Overlay environment variables (GEOTOOLS_ prefix, uppercase keys)
    env_map = {
        "GEOTOOLS_GDAL_CACHEMAX":  "gdal_cachemax",
        "GEOTOOLS_OUTPUT_CRS":     "output_crs",
        "GEOTOOLS_CHUNK_SIZE":     "chunk_size",
        "GEOTOOLS_WORKERS":        "workers",
        "GEOTOOLS_LOG_FORMAT":     "log_format",
    }
    for env_key, field_name in env_map.items():
        if val := os.environ.get(env_key):
            data[field_name] = val

    # 4. Pydantic validates types and ranges; raises ValidationError on bad input
    return AppConfig(**data)

# cli/error_handler.py — structured logging + TTY-aware output
import json
import logging
import sys
import os

class StructuredFormatter(logging.Formatter):
    """Emit JSON records for machine consumers; plain text for TTYs."""

    def format(self, record: logging.LogRecord) -> str:
        if os.isatty(sys.stderr.fileno()):
            # Human-readable when attached to a terminal
            return f"{record.levelname}: {record.getMessage()}"
        # JSON for log aggregators (Loki, CloudWatch, etc.)
        payload = {
            "level":   record.levelname,
            "logger":  record.name,
            "message": record.getMessage(),
            "time":    self.formatTime(record),
        }
        if record.exc_info:
            payload["exc_info"] = self.formatException(record.exc_info)
        return json.dumps(payload)

def configure_logging(level: str = "INFO") -> None:
    handler = logging.StreamHandler(sys.stderr)
    handler.setFormatter(StructuredFormatter())
    logging.basicConfig(level=level, handlers=[handler])

def exit_with_crs_mismatch(src_epsg: int, expected_epsg: int) -> None:
    logging.error(
        "CRS mismatch: source EPSG:%d, expected EPSG:%d",
        src_epsg, expected_epsg,
    )
    sys.exit(10)  # domain-specific exit code; triggers correct CI/CD branch

# tests/test_orchestration.py — mock engine, test orchestration
from pathlib import Path
from unittest.mock import patch, MagicMock
import pytest
from myapp.orchestration.clip import run_clip_job
from myapp.models import ClipConfig

@pytest.fixture
def clip_config(tmp_path: Path) -> ClipConfig:
    # Use in-memory paths; no real files needed for orchestration tests
    return ClipConfig(
        src=tmp_path / "input.tif",
        aoi=tmp_path / "aoi.geojson",
        dst=tmp_path / "output.tif",
        crs="EPSG:4326",
    )

def test_run_clip_job_calls_engine(clip_config: ClipConfig) -> None:
    with patch("myapp.orchestration.clip.clip_raster_to_aoi") as mock_engine:
        run_clip_job(clip_config)
    mock_engine.assert_called_once_with(
        clip_config.src, clip_config.aoi, ANY, clip_config.crs
    )

def test_run_clip_job_cleans_up_on_failure(clip_config: ClipConfig, tmp_path: Path) -> None:
    with patch(
        "myapp.orchestration.clip.clip_raster_to_aoi",
        side_effect=RuntimeError("engine failure"),
    ):
        with pytest.raises(RuntimeError):
            run_clip_job(clip_config)
    # Temporary file must be removed even when engine raises
    assert not any(tmp_path.glob("*.tif.tmp*"))

# tests/test_engine.py — use rasterio's in-memory driver, no disk I/O
import numpy as np
import rasterio
from rasterio.transform import from_bounds
from rasterio.crs import CRS
from pathlib import Path
from myapp.engine.windowed import apply_ndvi_pipeline

def make_in_memory_raster(path: str, crs: str = "EPSG:4326") -> None:
    """Write a tiny 4-band synthetic raster to /vsimem/ for fast tests."""
    transform = from_bounds(0, 0, 1, 1, width=64, height=64)
    with rasterio.open(
        path, "w",
        driver="GTiff", height=64, width=64, count=4,
        dtype="uint16", crs=CRS.from_epsg(4326), transform=transform,
    ) as dst:
        for band in range(1, 5):
            dst.write(np.random.randint(0, 3000, (64, 64), dtype="uint16"), band)

def test_ndvi_pipeline_produces_single_band(tmp_path: Path) -> None:
    src = tmp_path / "synthetic.tif"
    dst = tmp_path / "ndvi.tif"
    make_in_memory_raster(str(src))
    apply_ndvi_pipeline(src, dst)

    with rasterio.open(dst) as result:
        assert result.count == 1
        assert result.dtypes[0] == "float32"
        assert result.crs == CRS.from_epsg(4326)