Refactoring editor upload

Author: frodrigo

web_api/editor.py

@@ -1,5 +1,6 @@
 import io
-from typing import Dict, Optional
+import re
+from typing import Dict, List, Optional, Tuple
 from uuid import UUID
 
 import requests
@@ -63,32 +64,26 @@ async def save(
         session_data.changeset = changeset
         await backend.update(session_id, session_data)
 
-    # OsmChange
-    out = io.StringIO()
-    o = OsmSax.OsmSaxWriter(out, "UTF-8")
-    o.startDocument()
-    o.startElement("osmChange", {"version": "0.6", "generator": "OsmSax"})
-
-    methode = {"node": o.NodeCreate, "way": o.WayCreate, "relation": o.RelationCreate}
+    osmchange: List[Dict] = []
     for action in ("modify", "delete"):
         if action in json and len(json[action]) > 0:
-            o.startElement(action, {})
             for e in json[action]:
                 try:
                     ee = utils.fetch_osm_elem(e["type"], e["id"])
                 except Exception:
                     ee = None
                 if ee and ee["version"] == int(e["version"]):
-                    ee["changeset"] = changeset
-                    ee["tag"] = e["tags"]
-                    methode[e["type"]](ee)
+                    osmchange.append(
+                        {
+                            "action": action,
+                            "type": e["type"],
+                            "osm_elem": ee,
+                            "tags": e["tags"],
+                        }
+                    )
                 else:
                     # FIXME reject
                     pass
-            o.endElement(action)
-
-    o.endElement("osmChange")
-    osmchange = out.getvalue()
 
     # Fire the changeset
     _changeset_upload(session_data.oauth2_token, changeset, osmchange)
@@ -143,13 +138,71 @@ def _changeset_close(oauth2_token: str, id: str) -> None:
     request.raise_for_status()
 
 
-def _changeset_upload(oauth2_token: str, id: str, osmchange) -> None:
-    request = requests.post(
-        utils.remote_url_write + "api/0.6/changeset/" + id + "/upload",
-        data=osmchange.encode(),
-        headers={
-            "Authorization": f"Bearer {oauth2_token}",
-            "Content-Type": "application/xml; charset=utf-8",
-        },
-    )
-    request.raise_for_status()
+def _build_osmchange(changeset_id: str, elements: List[Dict]) -> str:
+    out = io.StringIO()
+    o = OsmSax.OsmSaxWriter(out, "UTF-8")
+    o.startDocument()
+    o.startElement("osmChange", {"version": "0.6", "generator": "OsmSax"})
+
+    methode = {"node": o.NodeCreate, "way": o.WayCreate, "relation": o.RelationCreate}
+    for action in ("modify", "delete"):
+        action_elems = [el for el in elements if el["action"] == action]
+        if action_elems:
+            o.startElement(action, {})
+            for el in action_elems:
+                ee = dict(el["osm_elem"])
+                ee["changeset"] = changeset_id
+                ee["tag"] = el["tags"]
+                methode[el["type"]](ee)
+            o.endElement(action)
+
+    o.endElement("osmChange")
+    return out.getvalue()
+
+
+def _parse_conflict_element(body: str) -> Tuple[Optional[str], Optional[int]]:
+    # Typical body: "Version mismatch: Provided 3, server had: 4 of Node 12345"
+    match = re.search(r"of (Node|Way|Relation) (\d+)", body, re.IGNORECASE)
+    if match:
+        return match.group(1).lower(), int(match.group(2))
+    return None, None
+
+
+def _changeset_upload(
+    oauth2_token: str, changeset_id: str, elements: List[Dict]
+) -> None:
+    while True:  # Do as much retries as we can
+        osmchange = _build_osmchange(changeset_id, elements)
+        response = requests.post(
+            utils.remote_url_write + "api/0.6/changeset/" + changeset_id + "/upload",
+            data=osmchange.encode(),
+            headers={
+                "Authorization": f"Bearer {oauth2_token}",
+                "Content-Type": "application/xml; charset=utf-8",
+            },
+        )
+
+        if response.status_code == 200:
+            break
+
+        if response.status_code != 409:
+            response.raise_for_status()
+
+        elem_type, elem_id = _parse_conflict_element(response.text)
+
+        if elem_type is None or elem_id is None:
+            # Cannot parse the conflict; raise so the caller sees the error.
+            response.raise_for_status()
+
+        elements = list(
+            filter(
+                lambda el: not (
+                    el["type"] == elem_type and el["osm_elem"]["id"] == elem_id
+                ),
+                elements,
+            )
+        )
+
+        # If no elements remain there is nothing left to upload.
+        if len(elements) == 0:
+            break