osmapi.changeset

  1"""
  2Changeset operations for the OpenStreetMap API.
  3"""
  4
  5import re
  6import urllib.parse
  7import xml.dom.minidom
  8import xml.parsers.expat
  9from contextlib import contextmanager
 10from collections.abc import Generator
 11from typing import Any, TYPE_CHECKING, cast
 12from xml.dom.minidom import Element
 13
 14from . import dom, errors, xmlbuilder, parser
 15
 16if TYPE_CHECKING:
 17    from .OsmApi import OsmApi
 18
 19
 20class ChangesetMixin:
 21    """Mixin providing changeset-related operations with pythonic method names."""
 22
 23    @contextmanager
 24    def changeset(
 25        self: "OsmApi", changeset_tags: dict[str, str] | None = None
 26    ) -> Generator[int, None, None]:
 27        """
 28        Context manager for a Changeset.
 29
 30        It opens a Changeset, uploads the changes and closes the changeset
 31        when used with the `with` statement:
 32
 33            #!python
 34            import osmapi
 35
 36            with api.changeset({"comment": "Import script XYZ"}) as changeset_id:
 37                print(f"Part of changeset {changeset_id}")
 38                api.node_create({"lon":1, "lat":1, "tag": {}})
 39
 40        If `changeset_tags` are given, this tags are applied (key/value).
 41
 42        Returns `changeset_id`
 43
 44        The changeset is closed on the way out even if the body raises, so
 45        that an error does not leave a changeset dangling and block the next
 46        one with an `OsmApi.ChangesetAlreadyOpenError`.
 47
 48        If no session is provided to authenticate the request,
 49        `OsmApi.AuthenticationMissingError` is raised.
 50
 51        If there is already an open changeset,
 52        `OsmApi.ChangesetAlreadyOpenError` is raised.
 53        """
 54        if changeset_tags is None:
 55            changeset_tags = {}
 56        # Create a new changeset
 57        changeset_id = self.changeset_create(changeset_tags)
 58        try:
 59            yield changeset_id
 60        finally:
 61            self.changeset_close()
 62
 63    def changeset_get(
 64        self: "OsmApi", changeset_id: int, include_discussion: bool = False
 65    ) -> dict[str, Any]:
 66        """
 67        Returns changeset with `changeset_id` as a dict.
 68
 69        `changeset_id` is the unique identifier of a changeset.
 70
 71        If `include_discussion` is set to `True` the changeset discussion
 72        will be available in the result.
 73        """
 74        path = f"/api/0.6/changeset/{changeset_id}"
 75        if include_discussion:
 76            path = f"{path}?include_discussion=true"
 77        data = self._session._get(path)
 78        changeset = cast(
 79            Element, dom.OsmResponseToDom(data, tag="changeset", single=True)
 80        )
 81        return dom.dom_parse_changeset(changeset, include_discussion=include_discussion)
 82
 83    def changeset_update(
 84        self: "OsmApi", changeset_tags: dict[str, str] | None = None
 85    ) -> int:
 86        """
 87        Updates current changeset with `changeset_tags`.
 88
 89        If no session is provided to authenticate the request,
 90        `OsmApi.AuthenticationMissingError` is raised.
 91
 92        If there is no open changeset,
 93        `OsmApi.NoChangesetOpenError` is raised.
 94
 95        If the changeset is already closed,
 96        `OsmApi.ChangesetClosedApiError` is raised.
 97        """
 98        if changeset_tags is None:
 99            changeset_tags = {}
100        if not self._current_changeset_id:
101            raise errors.NoChangesetOpenError("No changeset currently opened")
102        if "created_by" not in changeset_tags:
103            changeset_tags["created_by"] = self._created_by
104        try:
105            self._session._put(
106                f"/api/0.6/changeset/{self._current_changeset_id}",
107                xmlbuilder._xml_build("changeset", {"tag": changeset_tags}, data=self),
108                return_value=False,
109            )
110        except errors.ApiError as e:
111            if e.status == 409:
112                raise errors.ChangesetClosedApiError(
113                    e.status, e.reason, e.payload
114                ) from e
115            else:
116                raise
117        return self._current_changeset_id
118
119    def changeset_create(
120        self: "OsmApi", changeset_tags: dict[str, str] | None = None
121    ) -> int:
122        """
123        Opens a changeset.
124
125        If `changeset_tags` are given, this tags are applied (key/value).
126
127        Returns `changeset_id`
128
129        If no session is provided to authenticate the request,
130        `OsmApi.AuthenticationMissingError` is raised.
131
132        If there is already an open changeset,
133        `OsmApi.ChangesetAlreadyOpenError` is raised.
134        """
135        if changeset_tags is None:
136            changeset_tags = {}
137        if self._current_changeset_id:
138            raise errors.ChangesetAlreadyOpenError("Changeset already opened")
139        if "created_by" not in changeset_tags:
140            changeset_tags["created_by"] = self._created_by
141
142        # check if someone tries to create a test changeset to PROD
143        if (
144            self._api == "https://www.openstreetmap.org"
145            and changeset_tags.get("comment") == "My first test"
146        ):
147            raise errors.OsmApiError(
148                "DO NOT CREATE test changesets on the production server"
149            )
150
151        result = self._session._put(
152            "/api/0.6/changeset/create",
153            xmlbuilder._xml_build("changeset", {"tag": changeset_tags}, data=self),
154        )
155        self._current_changeset_id = int(result)
156        return self._current_changeset_id
157
158    def changeset_close(self: "OsmApi") -> int:
159        """
160        Closes current changeset.
161
162        Returns `changeset_id`.
163
164        If no session is provided to authenticate the request,
165        `OsmApi.AuthenticationMissingError` is raised.
166
167        If there is no open changeset,
168        `OsmApi.NoChangesetOpenError` is raised.
169
170        If the changeset is already closed,
171        `OsmApi.ChangesetClosedApiError` is raised.
172        """
173        if not self._current_changeset_id:
174            raise errors.NoChangesetOpenError("No changeset currently opened")
175        try:
176            self._session._put(
177                f"/api/0.6/changeset/{self._current_changeset_id}/close",
178                None,
179                return_value=False,
180            )
181            current_changeset_id = self._current_changeset_id
182            self._current_changeset_id = 0
183        except errors.ApiError as e:
184            if e.status == 409:
185                raise errors.ChangesetClosedApiError(
186                    e.status, e.reason, e.payload
187                ) from e
188            else:
189                raise
190        return current_changeset_id
191
192    def changeset_upload(
193        self: "OsmApi", changes_data: list[dict[str, Any]]
194    ) -> list[dict[str, Any]]:
195        """
196        Upload data with the `changes_data` list of dicts.
197
198        Returns list with updated ids.
199
200        If no session is provided to authenticate the request,
201        `OsmApi.AuthenticationMissingError` is raised.
202
203        If the changeset is already closed,
204        `OsmApi.ChangesetClosedApiError` is raised.
205        """
206        data = ""
207        data += '<?xml version="1.0" encoding="UTF-8"?>\n'
208        data += '<osmChange version="0.6" generator="'
209        data += self._created_by + '">\n'
210        for change in changes_data:
211            data += "<" + change["action"] + ">\n"
212            change_data = change["data"]
213            data += self._add_changeset_data(change_data, change["type"])
214            data += "</" + change["action"] + ">\n"
215        data += "</osmChange>"
216        try:
217            response_data = self._session._post(
218                f"/api/0.6/changeset/{self._current_changeset_id}/upload",
219                data.encode("utf-8"),
220                forceAuth=True,
221            )
222        except errors.ApiError as e:
223            if e.status == 409 and re.search(
224                r"The changeset .* was closed at .*", e.payload_str
225            ):
226                raise errors.ChangesetClosedApiError(
227                    e.status, e.reason, e.payload
228                ) from e
229            else:
230                raise
231        try:
232            result_dom = xml.dom.minidom.parseString(response_data)
233            diff_result = result_dom.getElementsByTagName("diffResult")[0]
234            result_elements = [
235                x for x in diff_result.childNodes if x.nodeType == x.ELEMENT_NODE
236            ]
237        except (xml.parsers.expat.ExpatError, IndexError) as e:
238            raise errors.XmlResponseInvalidError(
239                f"The XML response from the OSM API is invalid: {e!r}"
240            ) from e
241
242        for change in changes_data:
243            if change["action"] == "delete":
244                for change_element in change["data"]:
245                    change_element.pop("version")
246            else:
247                self._assign_id_and_version(result_elements, change["data"])
248
249        return changes_data
250
251    def changeset_download(self: "OsmApi", changeset_id: int) -> list[dict[str, Any]]:
252        """
253        Download data from changeset `changeset_id`.
254
255        Returns list of dict with type, action, and data.
256        """
257        uri = f"/api/0.6/changeset/{changeset_id}/download"
258        data = self._session._get(uri)
259        return parser.parse_osc(data)
260
261    def changesets_get(  # noqa: C901
262        self: "OsmApi",
263        min_lon: float | None = None,
264        min_lat: float | None = None,
265        max_lon: float | None = None,
266        max_lat: float | None = None,
267        userid: int | None = None,
268        username: str | None = None,
269        closed_after: str | None = None,
270        created_before: str | None = None,
271        only_open: bool = False,
272        only_closed: bool = False,
273    ) -> dict[int, dict[str, Any]]:
274        """
275        Returns a dict with the id of the changeset as key matching all criteria.
276
277        All parameters are optional. The bounding box is only applied if all
278        four of `min_lon`, `min_lat`, `max_lon` and `max_lat` are given.
279
280        If only some of the bounding box values are given,
281        `ValueError` is raised.
282        """
283        uri = "/api/0.6/changesets"
284        params: dict[str, Any] = {}
285        bbox = (min_lon, min_lat, max_lon, max_lat)
286        if any(coord is not None for coord in bbox):
287            if any(coord is None for coord in bbox):
288                raise ValueError(
289                    "A bounding box needs all of min_lon, min_lat, max_lon "
290                    "and max_lat, got "
291                    f"min_lon={min_lon}, min_lat={min_lat}, "
292                    f"max_lon={max_lon}, max_lat={max_lat}"
293                )
294            params["bbox"] = ",".join(str(coord) for coord in bbox)
295        if userid:
296            params["user"] = userid
297        if username:
298            params["display_name"] = username
299        if closed_after and not created_before:
300            params["time"] = closed_after
301        if created_before:
302            if not closed_after:
303                closed_after = "1970-01-01T00:00:00Z"
304            params["time"] = f"{closed_after},{created_before}"
305        if only_open:
306            params["open"] = 1
307        if only_closed:
308            params["closed"] = 1
309
310        if params:
311            uri += "?" + urllib.parse.urlencode(params)
312
313        data = self._session._get(uri)
314        changesets = cast(list[Element], dom.OsmResponseToDom(data, tag="changeset"))
315        result: dict[int, dict[str, Any]] = {}
316        for cur_changeset in changesets:
317            tmp_cs = dom.dom_parse_changeset(cur_changeset)
318            result[tmp_cs["id"]] = tmp_cs
319        return result
320
321    def changeset_comment(
322        self: "OsmApi", changeset_id: int, comment: str
323    ) -> dict[str, Any]:
324        """
325        Adds a comment to the changeset `changeset_id`.
326
327        `comment` should be a string.
328
329        Returns the updated changeset data dict.
330
331        If no session is provided to authenticate the request,
332        `OsmApi.AuthenticationMissingError` is raised.
333
334        If the changeset is already closed,
335        `OsmApi.ChangesetClosedApiError` is raised.
336        """
337        params = urllib.parse.urlencode({"text": comment})
338        try:
339            data = self._session._post(
340                f"/api/0.6/changeset/{changeset_id}/comment",
341                params,
342                forceAuth=True,
343            )
344        except errors.ApiError as e:
345            if e.status == 409:
346                raise errors.ChangesetClosedApiError(
347                    e.status, e.reason, e.payload
348                ) from e
349            else:
350                raise
351        changeset = cast(
352            Element,
353            dom.OsmResponseToDom(data, tag="changeset", single=True),
354        )
355        return dom.dom_parse_changeset(changeset, include_discussion=False)
356
357    def changeset_subscribe(self: "OsmApi", changeset_id: int) -> dict[str, Any]:
358        """
359        Subscribe to the changeset `changeset_id`.
360
361        Returns the updated changeset data dict.
362
363        If no session is provided to authenticate the request,
364        `OsmApi.AuthenticationMissingError` is raised.
365
366        If already subscribed to this changeset,
367        `OsmApi.AlreadySubscribedApiError` is raised.
368        """
369        try:
370            data = self._session._post(
371                f"/api/0.6/changeset/{changeset_id}/subscribe",
372                None,
373                forceAuth=True,
374            )
375        except errors.ApiError as e:
376            if e.status == 409:
377                raise errors.AlreadySubscribedApiError(
378                    e.status, e.reason, e.payload
379                ) from e
380            else:
381                raise
382        changeset = cast(
383            Element,
384            dom.OsmResponseToDom(data, tag="changeset", single=True),
385        )
386        return dom.dom_parse_changeset(changeset, include_discussion=False)
387
388    def changeset_unsubscribe(self: "OsmApi", changeset_id: int) -> dict[str, Any]:
389        """
390        Unsubscribe from the changeset `changeset_id`.
391
392        Returns the updated changeset data dict.
393
394        If no session is provided to authenticate the request,
395        `OsmApi.AuthenticationMissingError` is raised.
396
397        If not subscribed to this changeset,
398        `OsmApi.NotSubscribedApiError` is raised.
399        """
400        try:
401            data = self._session._post(
402                f"/api/0.6/changeset/{changeset_id}/unsubscribe",
403                None,
404                forceAuth=True,
405            )
406        except errors.ApiError as e:
407            if e.status == 404:
408                raise errors.NotSubscribedApiError(e.status, e.reason, e.payload) from e
409            else:
410                raise
411        changeset = cast(
412            Element,
413            dom.OsmResponseToDom(data, tag="changeset", single=True),
414        )
415        return dom.dom_parse_changeset(changeset, include_discussion=False)