Structured JSON Logging for Geospatial Pipelines

Free-text log lines are hostile to auditing: they cannot be filtered, aggregated, or verified at scale. This guide shows how to configure Python’s standard logging module to emit one machine-parseable JSON object per event, attach a correlation id to every record in a batch, and produce per-batch structured entries that a compliance reviewer can query directly. The result plugs straight into the schema laid out in the parent module, Validation Log Design for Compliance Pipelines.

Prerequisites

Step-by-step

Step 1: Build a JSON formatter

Subclass logging.Formatter so that every emitted line is a complete JSON object. Merge any structured fields passed through the extra argument, and always include a UTC timestamp and the level.

import json
import logging
from datetime import datetime, timezone

RESERVED = set(logging.LogRecord("", 0, "", 0, "", (), None).__dict__)

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        # Fold any structured fields passed via logger.info(..., extra={...}).
        for key, value in record.__dict__.items():
            if key not in RESERVED and not key.startswith("_"):
                payload[key] = value
        return json.dumps(payload, sort_keys=True, default=str)

Using sort_keys=True makes each line canonical, which matters when records are later hashed for tamper-evidence.

Step 2: Attach the formatter to a handler

Wire the formatter into a handler and logger once, at pipeline startup. Disable propagation so records are not double-emitted by the root logger.

def configure_audit_logger(sink_path: str) -> logging.Logger:
    logger = logging.getLogger("compliance.audit")
    logger.setLevel(logging.INFO)
    logger.propagate = False
    handler = logging.FileHandler(sink_path)   # swap for an object-store handler in prod
    handler.setFormatter(JsonFormatter())
    logger.handlers.clear()                    # idempotent: avoid duplicate handlers
    logger.addHandler(handler)
    return logger

Step 3: Propagate a correlation id

A correlation id ties every record from one run together. Store it in a contextvars.ContextVar so it survives across function boundaries and worker threads without being threaded through every signature manually. A logging.Filter injects it into each record automatically.

import contextvars
import uuid

batch_id_var = contextvars.ContextVar("batch_id", default="unset")

class CorrelationFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.batch_id = batch_id_var.get()   # stamped onto every emitted record
        return True

def start_batch() -> str:
    batch_id = uuid.uuid4().hex
    batch_id_var.set(batch_id)
    return batch_id

Add the filter to the handler with handler.addFilter(CorrelationFilter()). Every subsequent record now carries batch_id with no extra arguments at the call site.

Step 4: Emit a per-batch structured record

At the end of each stage, emit a single aggregate record. Always read the CRS from the data and project to a metric CRS before any distance summary, so logged metrics are measured in meters rather than degrees.

import geopandas as gpd

def log_stage_summary(logger: logging.Logger, stage: str,
                      gdf: gpd.GeoDataFrame, metric_crs: str = "EPSG:32610") -> None:
    # Project to a linear CRS before summarizing any area metric.
    projected = gdf.to_crs(metric_crs) if gdf.crs else gdf
    logger.info(
        "stage complete",
        extra={
            "stage": stage,
            "epsg_in": gdf.crs.to_epsg() if gdf.crs else None,
            "epsg_metric": 32610,
            "feature_count": int(len(gdf)),
            "empty_geometry_count": int(projected.geometry.is_empty.sum()),
            "total_area_sqm": float(projected.geometry.area.sum()),
        },
    )

Verification

Confirm that each emitted line is valid JSON, carries the correlation id, and shares one batch_id across the run. Parse the sink back and assert structure rather than eyeballing it.

import json

with open("audit.log") as fh:
    records = [json.loads(line) for line in fh if line.strip()]

batch_ids = {r["batch_id"] for r in records}
assert len(batch_ids) == 1, f"expected one batch id, saw {batch_ids}"
assert all("stage" in r and "feature_count" in r for r in records)
print(f"{len(records)} records verified for batch {batch_ids.pop()}")

If every record parses and the assertions hold, the log is queryable and ready for the hash-chaining and retention steps described in the parent module.

Common Pitfalls

  • Logging geometry objects directly. Shapely geometries are not JSON-serializable and will bloat the log if coerced to WKT. Log counts, areas, and identifiers instead of raw geometry; the default=str fallback should be a safety net, not the plan.
  • Summarizing area in a geographic CRS. Calling .area on EPSG:4326 data yields square degrees, which are meaningless. Always reproject to a metric CRS first, as in Step 4, and record which EPSG the metric was computed in.
  • Duplicate handlers on re-import. Calling the configuration function twice stacks handlers and emits every line multiple times. Clear existing handlers first, as shown, so the logger is idempotent.

Frequently Asked Questions

Why use the standard logging module instead of just writing JSON files myself?

The standard library gives you handler routing, log levels, thread-safety, and filters for free. A correlation-id filter stamps every record automatically, and swapping a file sink for an object-store or syslog handler becomes a one-line change rather than a rewrite. Rolling your own file writer usually reinvents these features less reliably.

How do correlation ids work across multiprocessing worker pools?

A contextvars.ContextVar is inherited by child threads but not by separate processes. When you fan out with multiprocessing, pass the batch id explicitly to each worker and call the setter inside the worker before logging. Each process then stamps the same id, so records from all workers still collate into one run in the audit store.

Should I log every feature or only aggregates?

Log per-batch aggregates by default and reserve per-feature records for failures and quarantined geometries. At municipal scale, per-feature logging inflates storage and slows queries, while aggregate records plus targeted failure entries give auditors everything they need without the volume.

How does this connect to capturing coordinate system details?

The formatter here carries whatever fields you pass in extra, including EPSG codes. To record the full source-to-target transformation pipeline rather than just a single EPSG code, follow Capturing CRS Provenance in Validation Logs, which extends these same structured records with pyproj pipeline detail.