Add support for HTTP headers

overpy/__init__.py

@@ -1,7 +1,6 @@
 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
@@ -60,45 +59,45 @@ class Overpass:
     """
     Class to access the Overpass API
 
-    :param read_chunk_size: Max size of each chunk read from the server response
     :param url: Optional URL of the Overpass server. Defaults to http://overpass-api.de/api/interpreter
     :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)
     default_max_retry_count: ClassVar[int] = 0
 
-    #: Max size of each chunk read from the server response
-    default_read_chunk_size: ClassVar[int] = 4096
-
     #: Global time to wait between tries (Default: 1.0s)
     default_retry_timeout: ClassVar[float] = 1.0
 
     #: Default URL of the Overpass server
     default_url: ClassVar[str] = "http://overpass-api.de/api/interpreter"
 
+    default_headers: ClassVar[Dict[str, str]] = {
+        'User-Agent': 'python-overpy/{} (urllib)'.format(__version__),
+    }
+
     def __init__(
             self,
-            read_chunk_size: Optional[int] = None,
             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"<[^>]*?>")
-        if read_chunk_size is None:
-            read_chunk_size = self.default_read_chunk_size
-
-        #: The chunk size for this instance
-        self.read_chunk_size = read_chunk_size
 
         if max_retry_count is None:
             max_retry_count = self.default_max_retry_count
@@ -150,10 +149,9 @@ def query(self, query: Union[bytes, str]) -> "Result":
 
             response = b""
             try:
-                with urlopen(self.url, query) as f:
-                    f_read = partial(f.read, self.read_chunk_size)
-                    for data in iter(f_read, b""):
-                        response += data
+                req = Request(self.url, data=query, headers=self.headers)
+                with urlopen(req) as f:
+                    response = f.read()
             except HTTPError as exc:
                 f = exc
 

tests/test_request.py

@@ -128,15 +128,6 @@ def do_POST(self):
 
 
 class TestQuery:
-    def test_chunk_size(self):
-        url, server = new_server_thread(HandleResponseJSON)
-
-        api = overpy.Overpass(read_chunk_size=128)
-        api.url = url
-        result = api.query("[out:json];node(50.745,7.17,50.75,7.18);out;")
-        stop_server_thread(server)
-        assert len(result.nodes) > 0
-
     def test_overpass_syntax_error(self):
         url, server = new_server_thread(HandleOverpassBadRequest)