Building Stable Surrogate Keys for OSM Features Jump to heading#

from __future__ import annotations

import logging
import sqlite3
import uuid
from dataclasses import dataclass
from datetime import date

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logger = logging.getLogger("osm.identity.surrogate")

SCHEMA = """
CREATE TABLE IF NOT EXISTS feature (
    surrogate_key TEXT PRIMARY KEY,
    kind          TEXT NOT NULL,
    created_on    TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS mapping (
    surrogate_key TEXT NOT NULL,
    osm_type      TEXT NOT NULL,
    osm_id        INTEGER NOT NULL,
    osm_version   INTEGER NOT NULL,
    valid_from    TEXT NOT NULL,
    valid_to      TEXT,                    -- NULL means currently valid
    PRIMARY KEY (surrogate_key, osm_type, osm_id, valid_from)
);
CREATE TABLE IF NOT EXISTS lineage (
    parent_key TEXT NOT NULL,
    child_key  TEXT NOT NULL,
    relation   TEXT NOT NULL,              -- 'split' | 'merge'
    occurred_on TEXT NOT NULL,
    PRIMARY KEY (parent_key, child_key, occurred_on)
);
"""


@dataclass(frozen=True)
class Mapping:
    surrogate_key: str
    osm_type: str
    osm_id: int
    osm_version: int


def connect(path: str) -> sqlite3.Connection:
    conn = sqlite3.connect(path)
    conn.executescript(SCHEMA)
    return conn


def mint(conn: sqlite3.Connection, kind: str) -> str:
    """A new opaque key. Never derived from content, never reused."""
    key = f"{kind}-{uuid.uuid4()}"
    conn.execute("INSERT INTO feature (surrogate_key, kind, created_on) "
                 "VALUES (?, ?, ?)", (key, kind, date.today().isoformat()))
    return key


def bind(conn: sqlite3.Connection, m: Mapping, on: date | None = None) -> None:
    conn.execute(
        "INSERT OR IGNORE INTO mapping (surrogate_key, osm_type, osm_id,"
        " osm_version, valid_from) VALUES (?, ?, ?, ?, ?)",
        (m.surrogate_key, m.osm_type, m.osm_id, m.osm_version,
         (on or date.today()).isoformat()))


def close_mapping(conn: sqlite3.Connection, key: str, osm_type: str,
                  osm_id: int, on: date | None = None) -> None:
    conn.execute(
        "UPDATE mapping SET valid_to = ? WHERE surrogate_key = ? "
        "AND osm_type = ? AND osm_id = ? AND valid_to IS NULL",
        ((on or date.today()).isoformat(), key, osm_type, osm_id))


def current_objects(conn: sqlite3.Connection, key: str) -> list[tuple[str, int]]:
    return [(r[0], r[1]) for r in conn.execute(
        "SELECT osm_type, osm_id FROM mapping WHERE surrogate_key = ? "
        "AND valid_to IS NULL ORDER BY osm_type, osm_id", (key,))]


def apply_split(conn: sqlite3.Connection, key: str,
                new_objects: list[Mapping], on: date | None = None) -> None:
    """A way split: the SURROGATE key is unchanged, the mapping gains rows.

    This is the whole point of the indirection. Downstream consumers keyed on
    the surrogate see nothing; the feature simply now spans several OSM ways.
    """
    when = on or date.today()
    existing = set(current_objects(conn, key))
    for m in new_objects:
        if (m.osm_type, m.osm_id) not in existing:
            bind(conn, m, when)
    # Any previously mapped object no longer present has been superseded.
    incoming = {(m.osm_type, m.osm_id) for m in new_objects}
    for osm_type, osm_id in existing - incoming:
        close_mapping(conn, key, osm_type, osm_id, when)
    conn.commit()
    logger.info("split applied to %s: now maps to %d object(s)",
                key, len(current_objects(conn, key)))


def apply_merge(conn: sqlite3.Connection, keys: list[str],
                kind: str, objects: list[Mapping], on: date | None = None) -> str:
    """Two features became one. A NEW key is minted and lineage recorded."""
    when = on or date.today()
    new_key = mint(conn, kind)
    for m in objects:
        bind(conn, Mapping(new_key, m.osm_type, m.osm_id, m.osm_version), when)
    for old in keys:
        for osm_type, osm_id in current_objects(conn, old):
            close_mapping(conn, old, osm_type, osm_id, when)
        conn.execute("INSERT OR IGNORE INTO lineage (parent_key, child_key,"
                     " relation, occurred_on) VALUES (?, ?, 'merge', ?)",
                     (old, new_key, when.isoformat()))
    conn.commit()
    logger.info("merged %d key(s) into %s", len(keys), new_key)
    return new_key


if __name__ == "__main__":
    conn = connect(":memory:")
    key = mint(conn, "route")
    bind(conn, Mapping(key, "way", 4305800, 7))
    apply_split(conn, key, [Mapping(key, "way", 4305800, 8),
                            Mapping(key, "way", 999001, 1)])