Uploading an OSM Changeset from Python Jump to heading#

from __future__ import annotations

import logging
import os
import xml.etree.ElementTree as ET
from dataclasses import dataclass

import requests

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

# Point at the DEVELOPMENT API until an edit has been reviewed end to end.
API = os.environ.get("OSM_API", "https://master.apis.dev.openstreetmap.org/api/0.6")
TOKEN = os.environ["OSM_OAUTH_TOKEN"]
HEADERS = {
    "Authorization": f"Bearer {TOKEN}",
    "User-Agent": "osm-pipeline-example/1.0 (contact@example.org)",
}


class VersionConflict(RuntimeError):
    """Somebody edited an object after we read it. Re-read; never bump."""


@dataclass(frozen=True)
class TagEdit:
    osm_type: str            # "node" | "way" | "relation"
    osm_id: int
    set_tags: dict[str, str]     # keys to add or overwrite
    remove_tags: tuple[str, ...] = ()


def read_object(osm_type: str, osm_id: int) -> ET.Element:
    """Fetch an object's CURRENT full state. This is the conflict defence."""
    response = requests.get(f"{API}/{osm_type}/{osm_id}", headers=HEADERS, timeout=30)
    response.raise_for_status()
    root = ET.fromstring(response.content)
    element = root.find(osm_type)
    if element is None:
        raise RuntimeError(f"{osm_type}/{osm_id} not found")
    return element


def apply_tag_edit(element: ET.Element, edit: TagEdit) -> bool:
    """Mutate a copy of the CURRENT element. Returns False if nothing changed."""
    current = {t.get("k"): t.get("v") for t in element.findall("tag")}
    updated = dict(current)
    updated.update(edit.set_tags)
    for key in edit.remove_tags:
        updated.pop(key, None)
    if updated == current:
        return False

    # Rebuild the tag children from the FULL updated set — a modify replaces,
    # so every surviving tag must be present in the document we upload.
    for tag in element.findall("tag"):
        element.remove(tag)
    for key, value in sorted(updated.items()):
        ET.SubElement(element, "tag", {"k": key, "v": value})
    return True


def open_changeset(comment: str, source: str) -> int:
    root = ET.Element("osm")
    changeset = ET.SubElement(root, "changeset")
    for key, value in {
        "comment": comment,
        "source": source,
        "created_by": "osm-pipeline-example 1.0",
    }.items():
        ET.SubElement(changeset, "tag", {"k": key, "v": value})
    response = requests.put(f"{API}/changeset/create", headers=HEADERS,
                            data=ET.tostring(root), timeout=30)
    response.raise_for_status()
    changeset_id = int(response.text.strip())
    logger.info("opened changeset %d", changeset_id)
    return changeset_id


def build_osm_change(elements: list[ET.Element], changeset_id: int) -> bytes:
    root = ET.Element("osmChange", {"version": "0.6",
                                    "generator": "osm-pipeline-example"})
    modify = ET.SubElement(root, "modify")
    for element in elements:
        element.set("changeset", str(changeset_id))
        modify.append(element)
    return ET.tostring(root)


def upload(changeset_id: int, document: bytes) -> str:
    response = requests.post(f"{API}/changeset/{changeset_id}/upload",
                             headers=HEADERS, data=document, timeout=120)
    if response.status_code == 409:
        raise VersionConflict(response.text.strip())
    response.raise_for_status()
    return response.text


def close_changeset(changeset_id: int) -> None:
    requests.put(f"{API}/changeset/{changeset_id}/close",
                 headers=HEADERS, timeout=30)
    logger.info("closed changeset %d", changeset_id)


def run(edits: list[TagEdit], comment: str, source: str) -> None:
    # Read as late as possible: the read-to-upload window is the conflict window.
    staged: list[ET.Element] = []
    for edit in edits:
        element = read_object(edit.osm_type, edit.osm_id)
        if apply_tag_edit(element, edit):
            staged.append(element)
        else:
            logger.info("%s/%d already correct, skipping", edit.osm_type, edit.osm_id)

    if not staged:
        logger.info("nothing to do")
        return

    changeset_id = open_changeset(comment, source)
    try:
        upload(changeset_id, build_osm_change(staged, changeset_id))
        logger.info("uploaded %d object(s) in changeset %d", len(staged), changeset_id)
    except VersionConflict as exc:
        # Do NOT increment the version and retry: that overwrites the other edit.
        logger.error("conflict — re-read the named object and decide again: %s", exc)
        raise
    finally:
        close_changeset(changeset_id)


if __name__ == "__main__":
    run(
        edits=[TagEdit("node", 1234567, {"amenity": "pharmacy"}, ("shop",))],
        comment="Retag shop=chemist to amenity=pharmacy in <area>; see wiki page",
        source="Local survey, 2026-09",
    )