Skip to content

Commit

Permalink
http: add support for headers
Browse files Browse the repository at this point in the history
It is useful to be able to specify additional headers to send when
making overpass requests. For example a special user agent, or an API
token.

Signed-off-by: Frank Villaro-Dixon <[email protected]>
  • Loading branch information
Frankkkkk committed May 6, 2024
1 parent 29ddf0c commit 5fcc3b8
Showing 1 changed file with 12 additions and 3 deletions.
15 changes: 12 additions & 3 deletions overpy/__init__.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from datetime import datetime
from decimal import Decimal
from functools import partial
from urllib.request import urlopen
from urllib.request import urlopen, Request
from urllib.error import HTTPError
from xml.sax import handler, make_parser
import xml.etree.ElementTree
Expand Down Expand Up @@ -64,6 +64,7 @@ class Overpass:
:param xml_parser: The xml parser to use
:param max_retry_count: Max number of retries (Default: default_max_retry_count)
:param retry_timeout: Time to wait between tries (Default: default_retry_timeout)
:param headers: Dict of headers to send with the request
"""

#: Global max number of retries (Default: 0)
Expand All @@ -75,18 +76,25 @@ class Overpass:
#: Default URL of the Overpass server
default_url: ClassVar[str] = "http://overpass-api.de/api/interpreter"

default_headers: ClassVar[Dict[str, str]] = {}

def __init__(
self,
url: Optional[str] = None,
xml_parser: int = XML_PARSER_SAX,
max_retry_count: int = None,
retry_timeout: float = None):
retry_timeout: float = None,
headers: dict[str, str] = None):

#: URL to use for this instance
self.url = self.default_url
if url is not None:
self.url = url

self.headers = self.default_headers
if headers is not None:
self.headers = headers

self._regex_extract_error_msg = re.compile(br"\<p\>(?P<msg>\<strong\s.*?)\</p\>")
self._regex_remove_tag = re.compile(b"<[^>]*?>")

Expand Down Expand Up @@ -140,7 +148,8 @@ def query(self, query: Union[bytes, str]) -> "Result":

response = b""
try:
with urlopen(self.url, query) as f:
req = Request(self.url, data=query, headers=self.headers)
with urlopen(req) as f:
response = f.read()
except HTTPError as exc:
f = exc
Expand Down

0 comments on commit 5fcc3b8

Please sign in to comment.