Revert "fix tests, deprecate 3.6 related functions"

This reverts commit 0f376ec.

.gitignore

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

overpass/api.py

@@ -7,13 +7,15 @@
 import json
 import logging
 import re
-from datetime import datetime
+
+from datetime import datetime, timezone
 from io import StringIO
 
 import geojson
 import requests
 from shapely.geometry import Point, Polygon
 
+from overpass import dependency
 from .errors import (
     MultipleRequestsError,
     OverpassSyntaxError,
@@ -24,7 +26,7 @@
 )
 
 
-class API:
+class API(object):
     """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,8 +61,11 @@ def __init__(self, *args, **kwargs):
 
         if self.debug:
             # https://stackoverflow.com/a/16630836
-            import http.client as http_client
-
+            try:
+                import http.client as http_client
+            except ImportError:
+                # Python 2
+                import httplib as http_client
             http_client.HTTPConnection.debuglevel = 1
 
             # You must initialize logging,
@@ -71,9 +76,7 @@ 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
@@ -92,7 +95,7 @@ def get(
                 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 = self._strptime(date)
         # Construct full Overpass query
         if build:
             full_query = self._construct_ql_query(
@@ -135,6 +138,16 @@ def get(
         # construct geojson
         return self._as_geojson(response["elements"])
 
+    @staticmethod
+    def _strptime(date_string):
+        if dependency.Python.less_3_7():
+            dt = datetime.strptime(date_string, '%Y-%m-%dT%H:%M:%SZ')
+            kwargs = {k: getattr(dt, k) for k in ('year', 'month', 'day', 'hour', 'minute', 'second', 'microsecond')}
+            kwargs['tzinfo'] = timezone.utc
+            return datetime(**kwargs)
+
+        return datetime.strptime(date_string, '%Y-%m-%dT%H:%M:%S%z')
+
     @classmethod
     def _api_status(cls) -> dict:
         """
@@ -145,29 +158,23 @@ def _api_status(cls) -> 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_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)
-        )
+        waiting_re = re.compile(r'(?<=Slot available after: )[\d\-TZ:]{20}')
+        waiting_slots = tuple(cls._strptime(waiting_re.search(line).group())
+                              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")
-        )
-        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
+            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_datetimes = tuple(cls._strptime(slot[3]) for slot in running_slots)
 
         return {
             "available_slots": available_slots,
@@ -198,24 +205,16 @@ 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,
-        )
+        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,
-        )
+        warnings.warn('Call to deprecated function "Search", use "search" function instead', DeprecationWarning)
         return self.search(*args, **kwargs)
 
     # deprecation of upper case functions
@@ -233,8 +232,7 @@ 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(
@@ -270,7 +268,7 @@ def _get_from_overpass(self, query):
             elif self._status == 504:
                 raise ServerLoadError(self._timeout)
             raise UnknownOverpassError(
-                f"The request returned status code {self._status}"
+                "The request returned status code {code}".format(code=self._status)
             )
         else:
             r.encoding = "utf-8"
@@ -286,9 +284,7 @@ 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)
@@ -310,38 +306,26 @@ def _as_geojson(self, elements):
                 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:
@@ -350,21 +334,19 @@ 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)
 

overpass/dependency.py

@@ -0,0 +1,9 @@
+import sys
+
+
+class Python:
+    version = (sys.version_info.major, sys.version_info.minor)
+
+    @classmethod
+    def less_3_7(cls):
+        return cls.version < (3, 7)

requirements-dev.txt

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

setup.cfg

@@ -0,0 +1,2 @@
+[metadata]
+description-file=README.md

setup.py

@@ -0,0 +1,25 @@
+from setuptools import setup
+
+setup(
+    name="overpass",
+    packages=["overpass"],
+    version="0.7",
+    description="Python wrapper for the OpenStreetMap Overpass API",
+    long_description="See README.md",
+    author="Martijn van Exel",
+    author_email="[email protected]",
+    url="https://github.com/mvexel/overpass-api-python-wrapper",
+    license="Apache",
+    keywords=["openstreetmap", "overpass", "wrapper"],
+    classifiers=[
+        "License :: OSI Approved :: Apache Software License",
+        "Programming Language :: Python :: 3.6",
+        "Programming Language :: Python :: 3.7",
+        "Programming Language :: Python :: 3.8",
+        "Programming Language :: Python :: 3.9",
+        "Topic :: Scientific/Engineering :: GIS",
+        "Topic :: Utilities",
+    ],
+    install_requires=["requests>=2.3.0", "geojson>=1.0.9", "shapely>=1.6.4"],
+    extras_require={"test": ["pytest"]},
+)

tests/test_status.py

@@ -2,25 +2,24 @@
 
 import overpass
 
-RESPONSE_TEXT = b"""Connected as: 3268165505
+
+RESPONSE_TEXT = b'''Connected as: 3268165505
 Current time: 2021-09-03T14:40:17Z
 Rate limit: 2
 1 slots available now.
 Slot available after: 2021-09-03T14:41:37Z, in 80 seconds.
 Currently running queries (pid, space limit, time limit, start time):
-"""
+'''
 
 
 class TestSlots:
-    def setup_method(self):
+    def setup(self):
         self.api = overpass.API(debug=True)
         self.requests = None
 
-    def teardown_method(self):
+    def teardown(self):
         assert self.requests.get.called
-        assert self.requests.get.call_args.args == (
-            "https://overpass-api.de/api/status",
-        )
+        assert self.requests.get.call_args.args == ('https://overpass-api.de/api/status',)
         assert not self.requests.post.called
 
     def test_slots_available(self, requests):
@@ -39,6 +38,4 @@ def test_slots_waiting(self, requests):
         requests.response._content = RESPONSE_TEXT
         self.requests = requests
 
-        assert self.api.slots_waiting == (
-            datetime.datetime(2021, 9, 3, 14, 41, 37, tzinfo=datetime.timezone.utc),
-        )
+        assert self.api.slots_waiting == (datetime.datetime(2021, 9, 3, 14, 41, 37, tzinfo=datetime.timezone.utc),)