Validating an OSM Tag Mapping Config with Pydantic Jump to heading
A tag-mapping configuration is code that happens to be written in YAML, and it fails like code — except that without validation it fails three hours into a run, on the one element that reached the broken branch.
Prerequisites Jump to heading
Conceptual minimum Jump to heading
Validation happens at three levels, and only the first is what people usually mean by it.
Structural validation checks that the file parses and that each entry has the fields it should, with the right types. A typed model does this for free and catches typos in field names, which are the most common configuration error by a wide margin.
Cross-field validation checks relationships within one entry: a rule with a value list must also name a key; a rule producing a numeric output must declare a numeric type; a default must be of the declared type.
Semantic validation checks relationships between entries, and it is the level a schema cannot express. Two rules matching the same tag where the first shadows the second; a rule whose key never appears in the data; an output field no consumer reads; a priority order with ties.
The last level is where the real value is, because those are the failures that produce wrong output rather than an error. A shadowed rule does not crash; it silently never fires, and the features it was meant to classify get whatever the earlier rule said.
Runnable solution Jump to heading
from __future__ import annotations
import logging
from collections import defaultdict
from typing import Annotated, Literal
import yaml
from pydantic import BaseModel, Field, ValidationError, model_validator
logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logger = logging.getLogger("osm.mapping.validate")
# The fields the target schema actually has. A rule producing anything else
# writes a column nothing reads.
TARGET_FIELDS = {"feature_class", "name", "surface", "speed_kph", "lanes",
"access", "layer"}
class Rule(BaseModel):
model_config = {"extra": "forbid"} # a misspelled field is an error
key: str = Field(min_length=1)
values: list[str] | None = None # None means "any value"
output_field: str
output_value: str | None = None # None means "use the tag value"
output_type: Literal["string", "integer", "float", "boolean"] = "string"
priority: Annotated[int, Field(ge=0, le=1000)] = 100
@model_validator(mode="after")
def check_internally_consistent(self) -> "Rule":
if self.output_field not in TARGET_FIELDS:
raise ValueError(
f"output_field {self.output_field!r} is not in the target "
f"schema; known fields are {sorted(TARGET_FIELDS)}")
if self.output_type in {"integer", "float"} and self.output_value:
# A literal output with a numeric type must actually parse.
try:
float(self.output_value)
except ValueError as exc:
raise ValueError(f"output_value {self.output_value!r} is not "
f"{self.output_type}") from exc
if self.values is not None and not self.values:
raise ValueError("values is an empty list; omit it to match any "
"value, or list the values you mean")
return self
class Mapping(BaseModel):
model_config = {"extra": "forbid"}
name: str
rules: list[Rule] = Field(min_length=1)
@model_validator(mode="after")
def check_no_shadowing(self) -> "Mapping":
"""Semantic checks: the failures that produce wrong output, not errors."""
problems: list[str] = []
by_key: dict[str, list[Rule]] = defaultdict(list)
for rule in self.rules:
by_key[rule.key].append(rule)
for key, rules in by_key.items():
ordered = sorted(rules, key=lambda r: -r.priority)
catch_all_at: int | None = None
for index, rule in enumerate(ordered):
if rule.values is None:
if catch_all_at is None:
catch_all_at = index
elif catch_all_at is not None:
# A value-specific rule after a catch-all on the same key
# can never fire. It looks fine and silently does nothing.
problems.append(
f"rule for {key}={rule.values} is shadowed by an "
f"earlier catch-all on {key!r} at priority "
f"{ordered[catch_all_at].priority}")
# Two rules claiming the same value for the same key.
seen: dict[str, int] = {}
for rule in ordered:
for value in rule.values or ():
if value in seen:
problems.append(
f"{key}={value} is matched by two rules at "
f"priorities {seen[value]} and {rule.priority}")
seen[value] = rule.priority
# Ties in priority make the outcome depend on file order.
priorities = [r.priority for r in rules]
if len(set(priorities)) != len(priorities):
problems.append(
f"rules for {key!r} share a priority; the winner would "
f"depend on the order they appear in the file")
if problems:
raise ValueError("; ".join(problems))
return self
def load(path: str) -> Mapping:
raw = yaml.safe_load(open(path, encoding="utf-8"))
try:
mapping = Mapping.model_validate(raw)
except ValidationError as exc:
# Fail here, before a single element is read, with the entry named.
logger.error("configuration is invalid:\n%s", exc)
raise
logger.info("loaded %d rule(s) across %d key(s)", len(mapping.rules),
len({r.key for r in mapping.rules}))
return mapping
def coverage_report(mapping: Mapping, observed_keys: set[str]) -> None:
"""Which rules can never fire against this data?"""
configured = {r.key for r in mapping.rules}
unreachable = configured - observed_keys
unmapped = observed_keys - configured
if unreachable:
logger.warning("%d rule key(s) never appear in the data: %s",
len(unreachable), sorted(unreachable)[:10])
logger.info("%d observed key(s) have no rule", len(unmapped))
if __name__ == "__main__":
logger.info("validate at load time; a shadowed rule never raises at runtime")
Step-by-step walkthrough Jump to heading
- Forbid unknown fields. A misspelled field name is the commonest configuration error, and silently ignoring it produces a rule that behaves nothing like what was written.
- Validate output fields against the real schema. A rule writing to a column the target does not have is either a typo or a plan nobody implemented, and both are worth catching before the run.
- Check literal values against their declared type. A numeric output whose literal does not parse fails at the first matching element, which may be hours in.
- Reject an empty value list. Empty and absent mean opposite things — match nothing and match everything — and the ambiguity is worth refusing rather than resolving.
- Detect shadowing per key. A value-specific rule ordered after a catch-all on the same key can never fire, which is the archetypal silent failure in a priority-ordered mapping.
- Detect duplicate value claims. Two rules matching the same key and value means one of them is dead, and which one depends on ordering that may not be stable.
- Refuse tied priorities. Ties make the outcome depend on the order entries appear in the file, which is not a property anybody intends to rely on.
- Report coverage separately. Which configured keys never appear in the data, and which observed keys have no rule, are findings rather than errors and belong in a report.
Verification Jump to heading
- A misspelled field fails. Add one and confirm the load raises rather than proceeding.
- A shadowed rule fails. Order a specific rule after a catch-all on the same key and confirm the error names both.
- A tied priority fails. Give two rules for one key the same priority and confirm the error explains the consequence.
- The error names the entry. Every validation failure should identify which rule is at fault, not just that something is.
- Coverage reports, not fails. An unreachable key should produce a warning rather than blocking a run, since data varies by region.
Common errors and fixes Jump to heading
| Symptom | Root cause | One-line fix |
|---|---|---|
| Rule silently does nothing | Misspelled field ignored by the loader | Forbid unknown fields in the model |
| A class never appears in output | A rule shadowed by a catch-all | Detect value rules ordered after a catch-all per key |
| Output changes after reformatting | Priorities tied, order decides | Refuse duplicate priorities within a key |
| Run fails hours in | Literal value type checked lazily | Validate literals against the declared type at load |
| Column written that nothing reads | Output field not checked | Validate output fields against the target schema |
| Empty value list matches everything | Empty and absent conflated | Reject an empty list explicitly |
| Config valid, data unmapped | Coverage never reported | Report configured keys absent from the data |
Specification reference Jump to heading
Pydantic validates data against typed models, applying field constraints and then any model-level validators, and reports every failure with the path to the offending field. Configuring a model to forbid extra fields turns an unexpected key into a validation error rather than a silently ignored value. See the Pydantic documentation for model validators and the extra-field configuration.
Frequently Asked Questions Jump to heading
Why forbid unknown fields rather than ignoring them?
Because a misspelled field name is indistinguishable from an unknown one, and ignoring it produces a rule that silently behaves differently from what its author wrote. A configuration is code, and a typo in code should be an error. The cost is that adding a field to the model becomes necessary before using it, which is a small and appropriate friction.
What is rule shadowing and why does it matter so much?
It is a value-specific rule placed after a catch-all on the same key, so the catch-all always matches first and the specific rule never fires. It matters because nothing raises: the features the specific rule was meant to classify get the catch-all’s answer instead, and the output is complete, plausible and wrong. Detecting it requires comparing rules against each other, which no per-entry schema can do.
Why are tied priorities worth rejecting?
Because they make the outcome depend on the order entries appear in the file, which is not a property anybody intends to depend on. Somebody reorders the configuration for readability, the output changes, and nothing in the change explains why. Requiring distinct priorities within a key costs one number per rule and removes an entire class of inexplicable behaviour.
Should an unreachable rule fail the load?
No, report it. A rule whose key never appears in the current extract may be entirely correct for another region, and failing the load would mean maintaining a separate configuration per area. Reporting it tells you the rule is doing nothing here without preventing the run, and a rule unreachable across every region you process is then worth investigating.
Related Jump to heading
- Batch Attribute Mapping Strategies — the parent topic and the mapping model.
- Mapping OSM Tags to a Fixed Schema with YAML — the configuration this validates.
- Handling Missing Tags in OSM Data Pipelines — what a catch-all rule is usually for.
- Designing a Star Schema for OSM Features — the target schema output fields are checked against.
- Setting Quality Thresholds That Fail a Build — the same fail-early discipline applied to data.
Up one level: Batch Attribute Mapping Strategies.