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