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, Optional, Tuple, Union
 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        auth: Optional[Tuple[str, str]] = None,
 26        session: Optional[requests.Session] = None,
 27        timeout: int = 30,
 28    ) -> None:
 29        self._api = base_url
 30        self._created_by = created_by
 31        self._timeout = timeout
 32
 33        try:
 34            self._auth: Optional[Any] = auth
 35            if not auth and session.auth:  # type: ignore[union-attr]
 36                self._auth = session.auth  # type: ignore[union-attr]
 37        except AttributeError:
 38            pass
 39
 40        self._http_session = session
 41        self._session = self._get_http_session()
 42
 43    def close(self) -> None:
 44        if self._session:
 45            self._session.close()
 46
 47    def _http_request(  # noqa: C901
 48        self,
 49        method: str,
 50        path: str,
 51        auth: bool,
 52        send: Optional[Union[str, bytes]],
 53        return_value: bool = True,
 54        params: Optional[dict] = None,
 55    ) -> bytes:
 56        """
 57        Returns the response generated by an HTTP request.
 58
 59        `method` is a HTTP method to be executed
 60        with the request data. For example: 'GET' or 'POST'.
 61        `path` is the path to the requested resource relative to the
 62        base API address stored in self._api. Should start with a
 63        slash character to separate the URL.
 64        `auth` is a boolean indicating whether authentication should
 65        be preformed on this request.
 66        `send` contains additional data that might be sent in a
 67        request.
 68        `return_value` indicates wheter this request should return
 69        any data or not.
 70
 71        If the username or password is missing,
 72        `OsmApi.UsernamePasswordMissingError` is raised.
 73
 74        If the requested element has been deleted,
 75        `OsmApi.ElementDeletedApiError` is raised.
 76
 77        If the requested element can not be found,
 78        `OsmApi.ElementNotFoundApiError` is raised.
 79
 80        If the response status code indicates an error,
 81        `OsmApi.ApiError` is raised.
 82        """
 83        logger.debug(f"{datetime.datetime.now():%Y-%m-%d %H:%M:%S} {method} {path}")
 84
 85        # Add API base URL to path
 86        path = self._api + path
 87
 88        if auth and not self._auth:
 89            raise errors.UsernamePasswordMissingError("Username/Password missing")
 90
 91        try:
 92            response = self._session.request(
 93                method, path, data=send, timeout=self._timeout, params=params
 94            )
 95        except requests.exceptions.Timeout as e:
 96            raise errors.TimeoutApiError(
 97                0, f"Request timed out (timeout={self._timeout})", ""
 98            ) from e
 99        except requests.exceptions.ConnectionError as e:
100            raise errors.ConnectionApiError(0, f"Connection error: {str(e)}", "") from e
101        except requests.exceptions.RequestException as e:
102            raise errors.ApiError(0, str(e), "") from e
103
104        if response.status_code != 200:
105            payload = response.content.strip()
106            if response.status_code == 401:
107                raise errors.UnauthorizedApiError(
108                    response.status_code, response.reason, payload
109                )
110            if response.status_code == 404:
111                raise errors.ElementNotFoundApiError(
112                    response.status_code, response.reason, payload
113                )
114            elif response.status_code == 410:
115                raise errors.ElementDeletedApiError(
116                    response.status_code, response.reason, payload
117                )
118            raise errors.ApiError(response.status_code, response.reason, payload)
119        if return_value and not response.content:
120            raise errors.ResponseEmptyApiError(
121                response.status_code, response.reason, ""
122            )
123
124        logger.debug(f"{datetime.datetime.now():%Y-%m-%d %H:%M:%S} {method} {path}")
125        return response.content
126
127    def _http(  # type: ignore[return-value]  # noqa: C901
128        self,
129        cmd: str,
130        path: str,
131        auth: bool,
132        send: Optional[Union[str, bytes]],
133        return_value: bool = True,
134        params: Optional[dict] = None,
135    ) -> bytes:
136        for i in it.count(1):
137            try:
138                return self._http_request(
139                    cmd, path, auth, send, return_value=return_value, params=params
140                )
141            except errors.ApiError as e:
142                if e.status >= 500:
143                    if i == self.MAX_RETRY_LIMIT:
144                        raise
145                    if i != 1:
146                        self._sleep()
147                    self._session = self._get_http_session()
148                else:
149                    logger.debug("ApiError Exception occured")
150                    raise
151            except errors.UsernamePasswordMissingError:
152                raise
153            except Exception as e:
154                logger.exception("General exception occured")
155                if i == self.MAX_RETRY_LIMIT:
156                    if isinstance(e, errors.OsmApiError):
157                        raise
158                    raise errors.MaximumRetryLimitReachedError(
159                        f"Give up after {i} retries"
160                    ) from e
161                if i != 1:
162                    self._sleep()
163                self._session = self._get_http_session()
164
165    def _get_http_session(self) -> requests.Session:
166        """
167        Creates a requests session for connection pooling.
168        """
169        if self._http_session:
170            session = self._http_session
171        else:
172            session = requests.Session()
173
174        session.auth = self._auth
175        session.headers.update({"user-agent": self._created_by})
176        return session
177
178    def _sleep(self) -> None:
179        time.sleep(5)
180
181    def _get(self, path: str, params: Optional[dict] = None) -> bytes:
182        return self._http("GET", path, False, None, params=params)
183
184    def _put(
185        self, path: str, data: Optional[Union[str, bytes]], return_value: bool = True
186    ) -> bytes:
187        return self._http("PUT", path, True, data, return_value=return_value)
188
189    def _post(
190        self,
191        path: str,
192        data: Optional[Union[str, bytes]],
193        optionalAuth: bool = False,
194        forceAuth: bool = False,
195        params: Optional[dict] = None,
196    ) -> bytes:
197        # the Notes API allows certain POSTs by non-authenticated users
198        auth = optionalAuth and self._auth
199        if forceAuth:
200            auth = True
201        return self._http("POST", path, bool(auth), data, params=params)
202
203    def _delete(self, path: str, data: Optional[Union[str, bytes]]) -> bytes:
204        return self._http("DELETE", path, True, data)