improve unittests and fix strptime of status datetime for 3.6

.gitignore

@@ -11,4 +11,5 @@ eggs/
 pip-log.txt
 docs/_build/
 Pipfile.lock
-venv/
+venv/
+.vscode/

overpass/api.py

@@ -24,7 +24,7 @@
 )
 
 
-class API(object):
+class API:
     """A simple Python wrapper for the OpenStreetMap Overpass API.
 
     :param timeout: If a single number, the TCP connection timeout for the request. If a tuple
@@ -59,11 +59,8 @@ def __init__(self, *args, **kwargs):
 
         if self.debug:
             # https://stackoverflow.com/a/16630836
-            try:
-                import http.client as http_client
-            except ImportError:
-                # Python 2
-                import httplib as http_client
+            import http.client as http_client
+
             http_client.HTTPConnection.debuglevel = 1
 
             # You must initialize logging,
@@ -74,7 +71,9 @@ def __init__(self, *args, **kwargs):
             requests_log.setLevel(logging.DEBUG)
             requests_log.propagate = True
 
-    def get(self, query, responseformat="geojson", verbosity="body", build=True, date=''):
+    def get(
+        self, query, responseformat="geojson", verbosity="body", build=True, date=""
+    ):
         """Pass in an Overpass query in Overpass QL.
 
         :param query: the Overpass QL query to send to the endpoint
@@ -93,7 +92,7 @@ def get(self, query, responseformat="geojson", verbosity="body", build=True, dat
                 date = datetime.fromisoformat(date)
             except ValueError:
                 # The 'Z' in a standard overpass date will throw fromisoformat() off
-                date = datetime.strptime(date, '%Y-%m-%dT%H:%M:%SZ')
+                date = datetime.strptime(date, "%Y-%m-%dT%H:%M:%SZ")
         # Construct full Overpass query
         if build:
             full_query = self._construct_ql_query(
@@ -136,8 +135,8 @@ def get(self, query, responseformat="geojson", verbosity="body", build=True, dat
         # construct geojson
         return self._as_geojson(response["elements"])
 
-    @staticmethod
-    def _api_status() -> dict:
+    @classmethod
+    def _api_status(cls) -> dict:
         """
         :returns: dict describing the client's status with the API
         """
@@ -146,31 +145,28 @@ def _api_status() -> dict:
         r = requests.get(endpoint)
         lines = tuple(r.text.splitlines())
 
-        available_re = re.compile(r'\d(?= slots? available)')
+        available_re = re.compile(r"\d(?= slots? available)")
         available_slots = int(
             available_re.search(lines[3]).group()
             if available_re.search(lines[3])
             else 0
         )
 
-        waiting_re = re.compile(r'(?<=Slot available after: )[\d\-TZ:]{20}')
+        waiting_re = re.compile(r"(?<=Slot available after: )[\d\-TZ:]{20}")
         waiting_slots = tuple(
-            datetime.strptime(
-                waiting_re.search(line).group(), "%Y-%m-%dT%H:%M:%S%z"
-            )
-            for line in lines if waiting_re.search(line)
+            datetime.strptime(waiting_re.search(line).group(), "%Y-%m-%dT%H:%M:%S%z")
+            for line in lines
+            if waiting_re.search(line)
         )
 
         current_idx = next(
-            i for i, word in enumerate(lines)
-            if word.startswith('Currently running queries')
+            i
+            for i, word in enumerate(lines)
+            if word.startswith("Currently running queries")
         )
-        running_slots = tuple(tuple(line.split()) for line in lines[current_idx + 1:])
+        running_slots = tuple(tuple(line.split()) for line in lines[current_idx + 1 :])
         running_slots_datetimes = tuple(
-            datetime.strptime(
-                slot[3], "%Y-%m-%dT%H:%M:%S%z"
-            )
-            for slot in running_slots
+            datetime.strptime(slot[3], "%Y-%m-%dT%H:%M:%S%z") for slot in running_slots
         )
 
         return {
@@ -202,11 +198,29 @@ def slots_running(self) -> tuple:
 
     def search(self, feature_type, regex=False):
         """Search for something."""
-        raise NotImplementedError()
+        raise NotImplementedError
+
+    def __deprecation_get(self, *args, **kwargs):
+        import warnings
+
+        warnings.warn(
+            'Call to deprecated function "Get", use "get" function instead',
+            DeprecationWarning,
+        )
+        return self.get(*args, **kwargs)
+
+    def __deprecation_search(self, *args, **kwargs):
+        import warnings
+
+        warnings.warn(
+            'Call to deprecated function "Search", use "search" function instead',
+            DeprecationWarning,
+        )
+        return self.search(*args, **kwargs)
 
     # deprecation of upper case functions
-    Get = get
-    Search = search
+    Get = __deprecation_get
+    Search = __deprecation_search
 
     def _construct_ql_query(self, userquery, responseformat, verbosity, date):
         raw_query = str(userquery).rstrip()
@@ -219,7 +233,8 @@ def _construct_ql_query(self, userquery, responseformat, verbosity, date):
         if responseformat == "geojson":
             template = self._GEOJSON_QUERY_TEMPLATE
             complete_query = template.format(
-                query=raw_query, verbosity=verbosity, date=date)
+                query=raw_query, verbosity=verbosity, date=date
+            )
         else:
             template = self._QUERY_TEMPLATE
             complete_query = template.format(
@@ -255,7 +270,7 @@ def _get_from_overpass(self, query):
             elif self._status == 504:
                 raise ServerLoadError(self._timeout)
             raise UnknownOverpassError(
-                "The request returned status code {code}".format(code=self._status)
+                f"The request returned status code {self._status}"
             )
         else:
             r.encoding = "utf-8"
@@ -271,7 +286,9 @@ def _as_geojson(self, elements):
                     continue
                 ids_already_seen.add(elem["id"])
             except KeyError:
-                raise UnknownOverpassError("Received corrupt data from Overpass (no id).")
+                raise UnknownOverpassError(
+                    "Received corrupt data from Overpass (no id)."
+                )
             elem_type = elem.get("type")
             elem_tags = elem.get("tags")
             elem_nodes = elem.get("nodes", None)
@@ -284,35 +301,47 @@ def _as_geojson(self, elements):
             if elem_user:
                 elem_tags["user"] = elem_user
             if elem_uid:
-                elem_tags["uid"] = elem_uid  
+                elem_tags["uid"] = elem_uid
             if elem_version:
-                elem_tags["version"] = elem_version                                                
+                elem_tags["version"] = elem_version
             elem_geom = elem.get("geometry", [])
             if elem_type == "node":
                 # Create Point geometry
                 geometry = geojson.Point((elem.get("lon"), elem.get("lat")))
             elif elem_type == "way":
                 # Create LineString geometry
-                geometry = geojson.LineString([(coords["lon"], coords["lat"]) for coords in elem_geom])
+                geometry = geojson.LineString(
+                    [(coords["lon"], coords["lat"]) for coords in elem_geom]
+                )
             elif elem_type == "relation":
                 # Initialize polygon list
                 polygons = []
                 # First obtain the outer polygons
                 for member in elem.get("members", []):
                     if member["role"] == "outer":
-                        points = [(coords["lon"], coords["lat"]) for coords in member.get("geometry", [])]
+                        points = [
+                            (coords["lon"], coords["lat"])
+                            for coords in member.get("geometry", [])
+                        ]
                         # Check that the outer polygon is complete
                         if points and points[-1] == points[0]:
                             polygons.append([points])
                         else:
-                            raise UnknownOverpassError("Received corrupt data from Overpass (incomplete polygon).")
+                            raise UnknownOverpassError(
+                                "Received corrupt data from Overpass (incomplete polygon)."
+                            )
                 # Then get the inner polygons
                 for member in elem.get("members", []):
                     if member["role"] == "inner":
-                        points = [(coords["lon"], coords["lat"]) for coords in member.get("geometry", [])]
+                        points = [
+                            (coords["lon"], coords["lat"])
+                            for coords in member.get("geometry", [])
+                        ]
                         # Check that the inner polygon is complete
                         if not points or points[-1] != points[0]:
-                            raise UnknownOverpassError("Received corrupt data from Overpass (incomplete polygon).")
+                            raise UnknownOverpassError(
+                                "Received corrupt data from Overpass (incomplete polygon)."
+                            )
                         # We need to check to which outer polygon the inner polygon belongs
                         point = Point(points[0])
                         for poly in polygons:
@@ -321,19 +350,21 @@ def _as_geojson(self, elements):
                                 poly.append(points)
                                 break
                         else:
-                            raise UnknownOverpassError("Received corrupt data from Overpass (inner polygon cannot "
-                                                       "be matched to outer polygon).")
+                            raise UnknownOverpassError(
+                                "Received corrupt data from Overpass (inner polygon cannot "
+                                "be matched to outer polygon)."
+                            )
                 # Finally create MultiPolygon geometry
                 if polygons:
                     geometry = geojson.MultiPolygon(polygons)
             else:
-                raise UnknownOverpassError("Received corrupt data from Overpass (invalid element).")
+                raise UnknownOverpassError(
+                    "Received corrupt data from Overpass (invalid element)."
+                )
 
             if geometry:
                 feature = geojson.Feature(
-                    id=elem["id"],
-                    geometry=geometry,
-                    properties=elem_tags
+                    id=elem["id"], geometry=geometry, properties=elem_tags
                 )
                 features.append(feature)
 

pyproject.toml

@@ -0,0 +1,50 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "overpass"
+dynamic = ["version"]
+description = "Python wrapper for the OpenStreetMap Overpass API"
+readme = "README.md"
+license.file = "LICENSE.txt"
+authors = [
+    { name = "Martijn van Exel", email = "[email protected]" },
+]
+keywords = [
+    "openstreetmap",
+    "overpass",
+    "wrapper",
+]
+classifiers = [
+    "License :: OSI Approved :: Apache Software License",
+    "Programming Language :: Python :: 3.7",
+    "Programming Language :: Python :: 3.8",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Topic :: Scientific/Engineering :: GIS",
+    "Topic :: Utilities",
+]
+dependencies = [
+    "geojson>=1.0.9",
+    "requests>=2.3.0",
+    "shapely>=1.6.4",
+]
+
+[project.optional-dependencies]
+test = [
+    "pytest"
+]
+
+[project.urls]
+Homepage = "https://github.com/mvexel/overpass-api-python-wrapper"
+
+[tool.hatch.version]
+path = "overpass/__init__.py"
+
+[tool.hatch.build.targets.sdist]
+include = [
+    "/overpass",
+]
+

requirements-dev.txt

@@ -1,2 +1,3 @@
-pytest>=6.2.0
+pytest>=6.2.5
 tox>=3.20.1
+mock>=4.0.3

tests/__init__.py

@@ -0,0 +1,35 @@
+import os
+
+# Set temporary True to update example.json/example.response files with real overpass endpoint data
+UPDATE_EXAMPLES = False
+
+RESOURCE_PATH = os.path.join(os.path.dirname(os.path.abspath(os.path.relpath(__file__))), 'resources')
+
+
+def save_resource(file_name, data):
+    with open(os.path.join(RESOURCE_PATH, file_name), 'wb') as f:
+        f.write(data)
+
+
+def load_resource(file_name):
+    with open(os.path.join(RESOURCE_PATH, file_name), 'rb') as f:
+        result = f.read()
+    return result
+
+
+def update_examples():
+    import pickle
+    import geojson
+    import overpass
+
+    query = 'rel(6518385);out body geom;way(10322303);out body geom;node(4927326183);'
+    api = overpass.API()
+    osm_geo = api.get(query, verbosity='body geom')
+    save_resource('example.response', pickle.dumps(api._get_from_overpass(f'[out:json];{query}out body geom;'),
+                                                   protocol=2))
+    save_resource('example.json', geojson.dumps(osm_geo).encode('utf8'))
+
+
+if UPDATE_EXAMPLES:
+    UPDATE_EXAMPLES = False
+    update_examples()

tests/conftest.py

@@ -0,0 +1,19 @@
+import pytest
+import mock
+
+import overpass
+
+
+@pytest.fixture(scope='function')
+def requests(monkeypatch):
+    mocker = mock.MagicMock()
+    mocker.response = overpass.api.requests.Response()
+
+    mocker.response.status_code = 200
+
+    mocker.get.return_value = mocker.response
+    mocker.post.return_value = mocker.response
+
+    monkeypatch.setattr(overpass.api, 'requests', mocker)
+
+    yield mocker