osmapi.http

  1"""
  2HTTP session management for the OpenStreetMap API.
  3"""
  4
  5import datetime
  6import itertools as it
  7import logging
  8import requests
  9import time
 10from typing import Any
 11
 12from . import errors
 13
 14logger = logging.getLogger(__name__)
 15
 16
 17class OsmApiSession:
 18    MAX_RETRY_LIMIT = 5
 19    """Maximum retries if a call to the remote API fails (default: 5)"""
 20
 21    def __init__(
 22        self,
 23        base_url: str,
 24        created_by: str,
 25        session: requests.Session | None = None,
 26        timeout: int = 30,
 27    ) -> None:
 28        self._api = base_url
 29        self._created_by = created_by
 30        self._timeout = timeout
 31
 32        # authentication is taken from the session (e.g. an OAuth 2.0 session)
 33        self._auth: Any = getattr(session, "auth", None)
 34
 35        # A caller-provided session can carry credentials in ways that are not
 36        # visible from the outside: `session.auth`, an `Authorization` header,
 37        # a custom transport adapter, or a `Session` subclass adding the token
 38        # per request (this is what requests-oauthlib does). Sniffing
 39        # `session.auth` therefore both rejects valid setups and accepts
 40        # sessions without any token, so it is not used to make this decision:
 41        # the only case in which authentication is certainly missing is a
 42        # session that was built here, i.e. one the caller didn't provide.
 43        # Everything else is sent, and the API answers with a 401 ->
 44        # `UnauthorizedApiError` if the authorization really was missing.
 45        self._can_authenticate: bool = session is not None
 46
 47        self._http_session = session
 48        self._session = self._get_http_session()
 49
 50    def close(self) -> None:
 51        if self._session:
 52            self._session.close()
 53
 54    def _http_request(  # noqa: C901
 55        self,
 56        method: str,
 57        path: str,
 58        auth: bool,
 59        send: str | bytes | None,
 60        return_value: bool = True,
 61        params: dict | None = None,
 62    ) -> bytes:
 63        """
 64        Returns the response generated by an HTTP request.
 65
 66        `method` is a HTTP method to be executed
 67        with the request data. For example: 'GET' or 'POST'.
 68        `path` is the path to the requested resource relative to the
 69        base API address stored in self._api. Should start with a
 70        slash character to separate the URL.
 71        `auth` is a boolean indicating whether authentication should
 72        be preformed on this request.
 73        `send` contains additional data that might be sent in a
 74        request.
 75        `return_value` indicates wheter this request should return
 76        any data or not.
 77
 78        If the request requires authentication and no session was provided to
 79        carry credentials, `OsmApi.AuthenticationMissingError` is raised. With
 80        a session, the request is sent and a rejected authorization surfaces
 81        as `OsmApi.UnauthorizedApiError` (HTTP 401).
 82
 83        If the requested element has been deleted,
 84        `OsmApi.ElementDeletedApiError` is raised.
 85
 86        If the requested element can not be found,
 87        `OsmApi.ElementNotFoundApiError` is raised.
 88
 89        If the response status code indicates an error,
 90        `OsmApi.ApiError` is raised.
 91        """
 92        logger.debug(f"{datetime.datetime.now():%Y-%m-%d %H:%M:%S} {method} {path}")
 93
 94        # Add API base URL to path
 95        path = self._api + path
 96
 97        if auth and not self._can_authenticate:
 98            raise errors.AuthenticationMissingError(
 99                "Authentication missing, this request requires an "
100                "authenticated session, but no session was provided "
101                "(see the OAuth 2.0 examples)"
102            )
103
104        try:
105            response = self._session.request(
106                method, path, data=send, timeout=self._timeout, params=params
107            )
108        except requests.exceptions.Timeout as e:
109            raise errors.TimeoutApiError(
110                0, f"Request timed out (timeout={self._timeout})", ""
111            ) from e
112        except requests.exceptions.ConnectionError as e:
113            raise errors.ConnectionApiError(0, f"Connection error: {str(e)}", "") from e
114        except requests.exceptions.RequestException as e:
115            raise errors.ApiError(0, str(e), "") from e
116
117        if response.status_code != 200:
118            payload = response.content.strip()
119            if response.status_code == 401:
120                raise errors.UnauthorizedApiError(
121                    response.status_code, response.reason, payload
122                )
123            if response.status_code == 404:
124                raise errors.ElementNotFoundApiError(
125                    response.status_code, response.reason, payload
126                )
127            elif response.status_code == 410:
128                raise errors.ElementDeletedApiError(
129                    response.status_code, response.reason, payload
130                )
131            raise errors.ApiError(response.status_code, response.reason, payload)
132        if return_value and not response.content:
133            raise errors.ResponseEmptyApiError(
134                response.status_code, response.reason, ""
135            )
136
137        logger.debug(f"{datetime.datetime.now():%Y-%m-%d %H:%M:%S} {method} {path}")
138        return response.content
139
140    def _http(  # type: ignore[return-value]  # noqa: C901
141        self,
142        cmd: str,
143        path: str,
144        auth: bool,
145        send: str | bytes | None,
146        return_value: bool = True,
147        params: dict | None = None,
148    ) -> bytes:
149        for i in it.count(1):
150            try:
151                return self._http_request(
152                    cmd, path, auth, send, return_value=return_value, params=params
153                )
154            except errors.ApiError as e:
155                if e.status >= 500:
156                    if i == self.MAX_RETRY_LIMIT:
157                        raise
158                    if i != 1:
159                        self._sleep()
160                    self._session = self._get_http_session()
161                else:
162                    logger.debug("ApiError Exception occured")
163                    raise
164            except errors.AuthenticationMissingError:
165                raise
166            except Exception as e:
167                logger.exception("General exception occured")
168                if i == self.MAX_RETRY_LIMIT:
169                    if isinstance(e, errors.OsmApiError):
170                        raise
171                    raise errors.MaximumRetryLimitReachedError(
172                        f"Give up after {i} retries"
173                    ) from e
174                if i != 1:
175                    self._sleep()
176                self._session = self._get_http_session()
177
178    def _get_http_session(self) -> requests.Session:
179        """
180        Creates a requests session for connection pooling.
181        """
182        if self._http_session:
183            session = self._http_session
184        else:
185            session = requests.Session()
186
187        session.auth = self._auth
188        session.headers.update({"user-agent": self._created_by})
189        return session
190
191    def _sleep(self) -> None:
192        time.sleep(5)
193
194    def _get(self, path: str, params: dict | None = None) -> bytes:
195        return self._http("GET", path, False, None, params=params)
196
197    def _put(
198        self, path: str, data: str | bytes | None, return_value: bool = True
199    ) -> bytes:
200        return self._http("PUT", path, True, data, return_value=return_value)
201
202    def _post(
203        self,
204        path: str,
205        data: str | bytes | None,
206        optionalAuth: bool = False,
207        forceAuth: bool = False,
208        params: dict | None = None,
209    ) -> bytes:
210        # the Notes API allows certain POSTs by non-authenticated users
211        auth = optionalAuth and self._can_authenticate
212        if forceAuth:
213            auth = True
214        return self._http("POST", path, bool(auth), data, params=params)
215
216    def _delete(self, path: str, data: str | bytes | None) -> bytes:
217        return self._http("DELETE", path, True, data)