osmapi.OsmApi

  1"""
  2The OsmApi module is a wrapper for the OpenStreetMap API.
  3As such it provides an easy access to the functionality of the API.
  4
  5You can find this module [on PyPI](https://pypi.python.org/pypi/osmapi)
  6or [on GitHub](https://github.com/metaodi/osmapi).
  7
  8Find all information about changes of the different versions of this module
  9[in the CHANGELOG](https://github.com/metaodi/osmapi/blob/master/CHANGELOG.md).
 10
 11
 12## Notes:
 13
 14* **dictionary keys** are _unicode_
 15* **changeset** is _integer_
 16* **version** is _integer_
 17* **tag** is a _dictionary_
 18* **timestamp** is _unicode_
 19* **user** is _unicode_
 20* **uid** is _integer_
 21* node **lat** and **lon** are _floats_
 22* way **nd** is list of _integers_
 23* relation **member** is a _list of dictionaries_ like
 24`{"role": "", "ref":123, "type": "node"}`
 25* All method names are in snake_case. The deprecated CamelCase versions
 26(e.g. `NodeGet`) were removed in version 6.0.
 27"""
 28
 29import re
 30import logging
 31from typing import Any, NoReturn
 32from xml.dom.minidom import Element
 33import requests
 34
 35from osmapi import __version__
 36from . import errors
 37from . import http
 38from . import xmlbuilder
 39from .node import NodeMixin
 40from .way import WayMixin
 41from .relation import RelationMixin
 42from .changeset import ChangesetMixin
 43from .note import NoteMixin
 44from .capabilities import CapabilitiesMixin
 45
 46logger = logging.getLogger(__name__)
 47
 48
 49class OsmApi(
 50    NodeMixin,
 51    WayMixin,
 52    RelationMixin,
 53    ChangesetMixin,
 54    NoteMixin,
 55    CapabilitiesMixin,
 56):
 57    """
 58    Main class of osmapi, instanciate this class to use osmapi
 59    """
 60
 61    def __init__(
 62        self,
 63        appid: str = "",
 64        created_by: str = f"osmapi/{__version__}",
 65        api: str = "https://www.openstreetmap.org",
 66        session: requests.Session | None = None,
 67        timeout: int = 30,
 68    ) -> None:
 69        """
 70        Initialized the OsmApi object.
 71
 72        To make authenticated requests (i.e. anything that writes to OSM),
 73        pass an authenticated `requests.Session` as the `session` parameter,
 74        see below. Username/password authentication was shut down by
 75        OpenStreetMap in July 2024 and the corresponding parameters
 76        (`username`, `password` and `passwordfile`) were removed in
 77        version 6.0 of osmapi, use OAuth 2.0 instead.
 78
 79        To credit the application that supplies changes to OSM, an `appid`
 80        can be provided.  This is a string identifying the application.
 81        If this is omitted "osmapi" is used.
 82
 83        It is possible to configure the URL to connect to using the `api`
 84        parameter.  By default this is the SSL version of the production API
 85        of OpenStreetMap, for testing purposes, one might prefer the official
 86        test instance at "api06.dev.openstreetmap.org" or any other valid
 87        OSM-API. To use an encrypted connection (HTTPS) simply add 'https://'
 88        in front of the hostname of the `api` parameter (e.g.
 89        https://api.openstreetmap.com).
 90
 91        The `session` parameter can be used to provide a custom requests
 92        http session object (requests.Session). This is how authentication
 93        is provided: any session that authenticates its requests works, be it
 94        by `session.auth`, by an `Authorization: Bearer` header, by a custom
 95        adapter, or by a `Session` subclass that adds the token per request.
 96        Sessions are also useful for custom adapters, hooks etc.
 97
 98        Without a session, requests that require authentication raise
 99        `OsmApi.AuthenticationMissingError` before anything is sent. With a
100        session, they are sent as they are, and a missing or invalid
101        authorization is reported by the API as
102        `OsmApi.UnauthorizedApiError`.
103
104        Finally the `timeout` parameter is used by the http session to
105        throw an expcetion if the the timeout (in seconds) has passed without
106        an answer from the server.
107        """
108        # Get API
109        self._api: str = api.strip("/")
110
111        # Get created_by
112        if not appid:
113            self._created_by: str = created_by
114        else:
115            self._created_by = f"{appid} ({created_by})"
116
117        # Initialisation
118        self._current_changeset_id: int = 0
119
120        # Http connection
121        self.http_session: requests.Session | None = session
122        self._timeout: int = timeout
123        self._session: http.OsmApiSession = http.OsmApiSession(
124            self._api,
125            self._created_by,
126            session=self.http_session,
127            timeout=self._timeout,
128        )
129
130    def __enter__(self) -> "OsmApi":
131        self._session = http.OsmApiSession(
132            self._api,
133            self._created_by,
134            session=self.http_session,
135            timeout=self._timeout,
136        )
137        return self
138
139    def __exit__(self, *args: Any) -> None:
140        self.close()
141
142    def close(self) -> None:
143        if self._session:
144            self._session.close()
145
146    ##################################################
147    # Internal method                                #
148    ##################################################
149
150    def _raise_write_error(self, e: errors.ApiError) -> NoReturn:
151        """
152        Translate an `ApiError` raised by an element write into a typed error.
153
154        A 409 means either that the changeset has since been closed or that the
155        element version is out of date; a 412 means a precondition (usually a
156        referenced element) was not met. Anything else is re-raised unchanged.
157        """
158        if e.status == 409:
159            if re.search(r"The changeset .* was closed at .*", e.payload_str):
160                raise errors.ChangesetClosedApiError(
161                    e.status, e.reason, e.payload
162                ) from e
163            raise errors.VersionMismatchApiError(e.status, e.reason, e.payload) from e
164        elif e.status == 412:
165            raise errors.PreconditionFailedApiError(
166                e.status, e.reason, e.payload
167            ) from e
168        raise e
169
170    def _do(  # type: ignore[return-value]
171        self, action: str, osm_type: str, osm_data: dict[str, Any]
172    ) -> dict[str, Any]:
173        if not self._current_changeset_id:
174            raise errors.NoChangesetOpenError(
175                "You need to open a changeset before uploading data"
176            )
177        if "timestamp" in osm_data:
178            osm_data.pop("timestamp")
179        osm_data["changeset"] = self._current_changeset_id
180        if action == "create":
181            return self._do_create(osm_type, osm_data)
182        elif action == "modify":
183            return self._do_modify(osm_type, osm_data)
184        elif action == "delete":
185            return self._do_delete(osm_type, osm_data)
186
187    def _do_create(self, osm_type: str, osm_data: dict[str, Any]) -> dict[str, Any]:
188        if osm_data.get("id", -1) > 0:
189            raise errors.OsmTypeAlreadyExistsError(f"This {osm_type} already exists")
190        try:
191            result = self._session._put(
192                f"/api/0.6/{osm_type}/create",
193                xmlbuilder._xml_build(osm_type, osm_data, data=self),
194            )
195        except errors.ApiError as e:
196            self._raise_write_error(e)
197        osm_data["id"] = int(result.strip())
198        osm_data["version"] = 1
199        return osm_data
200
201    def _do_modify(self, osm_type: str, osm_data: dict[str, Any]) -> dict[str, Any]:
202        try:
203            result = self._session._put(
204                f"/api/0.6/{osm_type}/{osm_data['id']}",
205                xmlbuilder._xml_build(osm_type, osm_data, data=self),
206            )
207        except errors.ApiError as e:
208            logger.error(e.reason)
209            self._raise_write_error(e)
210        osm_data["version"] = int(result.strip())
211        return osm_data
212
213    def _do_delete(self, osm_type: str, osm_data: dict[str, Any]) -> dict[str, Any]:
214        try:
215            result = self._session._delete(
216                f"/api/0.6/{osm_type}/{osm_data['id']}",
217                xmlbuilder._xml_build(osm_type, osm_data, data=self),
218            )
219        except errors.ApiError as e:
220            self._raise_write_error(e)
221        osm_data["version"] = int(result.strip())
222        osm_data["visible"] = False
223        return osm_data
224
225    def _add_changeset_data(self, change_data: list[dict[str, Any]], type: str) -> str:
226        data = ""
227        for changed_element in change_data:
228            changed_element["changeset"] = self._current_changeset_id
229            xml_data = xmlbuilder._xml_build(type, changed_element, False, data=self)
230            data += xml_data.decode("utf-8")
231        return data
232
233    def _assign_id_and_version(
234        self, response_data: list[Element], request_data: list[dict[str, Any]]
235    ) -> None:
236        for response, element in zip(response_data, request_data):
237            element["id"] = int(response.getAttribute("new_id"))
238            element["version"] = int(response.getAttribute("new_version"))