How to Build a Typer CLI for Shapefile Conversion

pip install "typer[all]" geopandas pyogrio

#!/usr/bin/env python3
"""shapefile_converter.py — batch-convert ESRI Shapefiles to modern vector formats."""
from pathlib import Path
from typing import List, Optional

import typer
from rich.progress import track
import geopandas as gpd
from pyproj import CRS
from pyproj.exceptions import CRSError

app = typer.Typer(
    help="Batch-convert ESRI Shapefiles to GeoJSON, GeoPackage, FlatGeobuf, or Parquet.",
    add_completion=True,
)

SUPPORTED_FORMATS = {"geojson", "gpkg", "fgb", "parquet"}
DRIVER_MAP = {
    "geojson":  "GeoJSON",
    "gpkg":     "GPKG",
    "fgb":      "FlatGeobuf",
    "parquet":  "Parquet",        # written via gdf.to_parquet(), driver key unused
}


def resolve_targets(input_path: Path) -> List[Path]:
    """Normalise a single .shp file, directory, or glob into an explicit file list."""
    if input_path.is_file() and input_path.suffix.lower() == ".shp":
        return [input_path]
    if input_path.is_dir():
        return sorted(input_path.glob("*.shp"))
    # Treat the string as a glob relative to cwd
    resolved = sorted(Path.cwd().glob(str(input_path)))
    return [p for p in resolved if p.suffix.lower() == ".shp"]


def _validate_crs(crs_string: str) -> None:
    """Exit with code 1 if the CRS string is not accepted by pyproj."""
    try:
        CRS.from_user_input(crs_string)
    except CRSError as exc:
        typer.echo(f"Invalid CRS '{crs_string}': {exc}", err=True)
        raise typer.Exit(code=1)


@app.command()
def convert(
    input_path: Path = typer.Argument(..., help="Path to .shp file, directory, or glob"),
    output_format: str = typer.Option(
        "gpkg", "--output-format", "-f",
        help=f"Target format: {', '.join(sorted(SUPPORTED_FORMATS))}",
    ),
    target_crs: Optional[str] = typer.Option(
        None, "--target-crs",
        help="Reproject to this CRS before writing (e.g. EPSG:4326).",
    ),
    output_dir: Path = typer.Option(Path("converted"), "--output-dir", "-o"),
    quiet: bool = typer.Option(False, "--quiet", "-q", help="Suppress Rich progress bar."),
) -> None:
    # --- 1. Validate format early, before touching the filesystem ---
    fmt = output_format.lower()
    if fmt not in SUPPORTED_FORMATS:
        typer.echo(f"Unsupported format '{fmt}'. Choose from: {', '.join(sorted(SUPPORTED_FORMATS))}", err=True)
        raise typer.Exit(code=1)

    # --- 2. Validate CRS string before any file I/O ---
    if target_crs:
        _validate_crs(target_crs)

    # --- 3. Resolve input to an explicit list of .shp paths ---
    files = resolve_targets(input_path)
    if not files:
        typer.echo(f"No .shp files found matching: {input_path}", err=True)
        raise typer.Exit(code=2)

    output_dir.mkdir(parents=True, exist_ok=True)
    success_count = 0
    fail_count = 0

    iterator = files if quiet else track(files, description="Converting")

    for src in iterator:
        try:
            # --- 4. Read with pyogrio (thread-safe, zero-copy C layer over GDAL) ---
            gdf = gpd.read_file(src, engine="pyogrio")

            # --- 5. Reproject if requested (pyproj handles authority validation) ---
            if target_crs:
                gdf = gdf.to_crs(target_crs)

            out_path = output_dir / f"{src.stem}.{fmt}"

            # --- 6. Write: Parquet uses a dedicated method; all others use to_file() ---
            if fmt == "parquet":
                gdf.to_parquet(out_path)
            else:
                gdf.to_file(out_path, driver=DRIVER_MAP[fmt], engine="pyogrio")

            success_count += 1
        except Exception as exc:
            typer.echo(f"  FAILED {src.name}: {exc}", err=True)
            fail_count += 1

    typer.echo(f"\nDone — {success_count} converted, {fail_count} failed.")
    if fail_count > 0:
        raise typer.Exit(code=3)


if __name__ == "__main__":
    app()

if target_crs:
    if gdf.crs is None:
        typer.echo(
            f"  WARN {src.name}: no CRS defined — assuming EPSG:4326 before reprojection.",
            err=True,
        )
        gdf = gdf.set_crs("EPSG:4326")
    gdf = gdf.to_crs(target_crs)

# Quick check with ogrinfo (ships with GDAL)
ogrinfo -al -so converted/admin_boundaries.gpkg

# Programmatic check — row count and CRS authority code
python - <<'EOF'
import geopandas as gpd
gdf = gpd.read_file("converted/admin_boundaries.gpkg", engine="pyogrio")
assert len(gdf) > 0, "Output is empty"
assert gdf.crs is not None, "CRS is missing"
print(f"rows={len(gdf)}  crs={gdf.crs.to_epsg()}  geom={gdf.geom_type.unique()}")
EOF

rows=3247  crs=4326  geom=['MultiPolygon']