Processing 100k GeoJSON files with Python asyncio

#!/usr/bin/env python3
"""Async GeoJSON batch processor for 100k+ files."""
import asyncio
import json
import logging
import sys
from pathlib import Path
from typing import Optional

import aiofiles
from aiofiles.os import makedirs

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    stream=sys.stderr,
)
logger = logging.getLogger(__name__)

MAX_CONCURRENCY = 80
BATCH_SIZE = 500
MAX_RETRIES = 3

async def read_geojson(path: Path, semaphore: asyncio.Semaphore) -> Optional[dict]:
    """Read and parse a GeoJSON file with backpressure and retries."""
    async with semaphore:
        for attempt in range(MAX_RETRIES):
            try:
                async with aiofiles.open(path, mode="r", encoding="utf-8") as f:
                    content = await f.read()
                return json.loads(content)
            except (json.JSONDecodeError, UnicodeDecodeError) as exc:
                logger.warning("Parse error in %s (attempt %d): %s", path.name, attempt + 1, exc)
                return None
            except Exception as exc:
                delay = 0.1 * (2 ** attempt)
                logger.warning("Read error in %s (attempt %d): %s. Retrying in %.1fs", path.name, attempt + 1, exc, delay)
                await asyncio.sleep(delay)
        logger.error("Failed to read %s after %d retries", path.name, MAX_RETRIES)
        return None

def validate_and_transform(data: dict) -> Optional[dict]:
    """Validate GeoJSON structure and apply a lightweight coordinate transform."""
    if not isinstance(data, dict) or data.get("type") != "FeatureCollection":
        return None
    # Example: round coordinates to 6 decimal places
    for feature in data.get("features", []):
        geom = feature.get("geometry", {})
        if geom.get("type") == "Point" and "coordinates" in geom:
            geom["coordinates"] = [round(c, 6) for c in geom["coordinates"]]
    return data

async def batch_writer(output_dir: Path, results_queue: asyncio.Queue, batch_size: int = BATCH_SIZE):
    """Consume validated results and write them in batches."""
    await makedirs(str(output_dir), exist_ok=True)
    batch = []
    while True:
        item = await results_queue.get()
        if item is None:  # Sentinel value
            if batch:
                await _flush_batch(output_dir, batch)
            break
        batch.append(item)
        if len(batch) >= batch_size:
            await _flush_batch(output_dir, batch)
            batch.clear()

async def _flush_batch(output_dir: Path, batch: list):
    """Write a batch of GeoJSON objects to disk."""
    out_path = output_dir / f"batch_{id(batch)}.json"
    async with aiofiles.open(out_path, mode="w", encoding="utf-8") as f:
        await f.write(json.dumps({"type": "FeatureCollection", "features": batch}))
    logger.info("Wrote batch of %d features to %s", len(batch), out_path.name)

async def worker(path: Path, semaphore: asyncio.Semaphore, results_queue: asyncio.Queue):
    """Process a single file and push valid results to the queue."""
    raw = await read_geojson(path, semaphore)
    if raw is None:
        return
    transformed = validate_and_transform(raw)
    if transformed:
        await results_queue.put(transformed)

async def main(input_dir: Path, output_dir: Path):
    paths = list(input_dir.rglob("*.geojson"))
    logger.info("Discovered %d files. Starting pipeline...", len(paths))

    semaphore = asyncio.Semaphore(MAX_CONCURRENCY)
    results_queue = asyncio.Queue(maxsize=BATCH_SIZE * 2)

    # Start writer
    writer_task = asyncio.create_task(batch_writer(output_dir, results_queue))

    # Schedule workers
    tasks = [worker(p, semaphore, results_queue) for p in paths]
    try:
        await asyncio.gather(*tasks)
    finally:
        # Signal writer to finish
        await results_queue.put(None)
        await writer_task

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="Async GeoJSON batch processor")
    parser.add_argument("input_dir", type=Path)
    parser.add_argument("output_dir", type=Path)
    args = parser.parse_args()
    asyncio.run(main(args.input_dir, args.output_dir))