Skip to content

Commit

Permalink
fix tests, deprecate 3.6 related functions
Browse files Browse the repository at this point in the history
also migrated some metadata to pyproject.toml
  • Loading branch information
mvexel committed Feb 12, 2024
1 parent 55ab07c commit 0f376ec
Show file tree
Hide file tree
Showing 9 changed files with 131 additions and 94 deletions.
3 changes: 2 additions & 1 deletion .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -11,4 +11,5 @@ eggs/
pip-log.txt
docs/_build/
Pipfile.lock
venv/
venv/
.vscode/
110 changes: 64 additions & 46 deletions overpass/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,15 +7,13 @@
import json
import logging
import re

from datetime import datetime, timezone
from datetime import datetime
from io import StringIO

import geojson
import requests
from shapely.geometry import Point, Polygon

from overpass import dependency
from .errors import (
MultipleRequestsError,
OverpassSyntaxError,
Expand All @@ -26,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
Expand Down Expand Up @@ -61,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,
Expand All @@ -76,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
Expand All @@ -95,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 = self._strptime(date)
date = datetime.strptime(date, "%Y-%m-%dT%H:%M:%SZ")
# Construct full Overpass query
if build:
full_query = self._construct_ql_query(
Expand Down Expand Up @@ -138,16 +135,6 @@ def get(self, query, responseformat="geojson", verbosity="body", build=True, dat
# 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:
"""
Expand All @@ -158,23 +145,29 @@ 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(cls._strptime(waiting_re.search(line).group())
for line in lines if waiting_re.search(line))
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)
)

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_datetimes = tuple(
datetime.strptime(slot[3], "%Y-%m-%dT%H:%M:%S%z") for slot in running_slots
)
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,
Expand Down Expand Up @@ -205,16 +198,24 @@ 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
Expand All @@ -232,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(
Expand Down Expand Up @@ -268,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"
Expand All @@ -284,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)
Expand All @@ -306,26 +310,38 @@ 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:
Expand All @@ -334,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)

Expand Down
9 changes: 0 additions & 9 deletions overpass/dependency.py

This file was deleted.

50 changes: 50 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
@@ -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",
]

2 changes: 1 addition & 1 deletion requirements-dev.txt
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
pytest>=6.2.0
pytest>=6.2.5
tox>=3.20.1
mock>=4.0.3
2 changes: 0 additions & 2 deletions setup.cfg

This file was deleted.

25 changes: 0 additions & 25 deletions setup.py

This file was deleted.

17 changes: 10 additions & 7 deletions tests/test_status.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,24 +2,25 @@

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(self):
def setup_method(self):
self.api = overpass.API(debug=True)
self.requests = None

def teardown(self):
def teardown_method(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):
Expand All @@ -38,4 +39,6 @@ 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),
)
Loading

0 comments on commit 0f376ec

Please sign in to comment.