git - Merge pull request #107 from eumiro/modernize

Modernize code

Author: phibos

README.rst

@@ -33,7 +33,7 @@ Install
 
 Supported Python versions:
 
-* Python >= 3.6.2
+* Python >= 3.7
 * PyPy3
 
 **Install:**

docs/source/introduction.rst

@@ -6,7 +6,7 @@ Requirements
 
 Supported Python versions:
 
-* Python 3.6+
+* Python 3.7+
 * PyPy3
 
 Installation

examples/get_nodes.py

@@ -8,15 +8,9 @@
 result = api.query(f"node({jet_deau}); out meta;")
 node = result.get_node(jet_deau)
 
-print(
- "The node for the famous Geneva {} ({},{}) was:".format(
- node.tags['name'],
- node.lat,
- node.lon
- )
-)
+print(f"The node for the famous Geneva {node.tags['name']} ({node.lat},{node.lon}) was:")
 attrs = node.attributes
 
-print("* last modified {}".format(attrs['timestamp']))
-print("* by {} (uid: {})".format(attrs['user'], attrs['uid']))
-print("* in changeset {}".format(attrs['changeset']))
+print(f"* last modified {attrs['timestamp']}")
+print(f"* by {attrs['user']} (uid: {attrs['uid']})")
+print(f"* in changeset {attrs['changeset']}")

examples/get_ways.py

@@ -11,8 +11,8 @@
  """)
 
 for way in result.ways:
- print("Name: %s" % way.tags.get("name", "n/a"))
- print(" Highway: %s" % way.tags.get("highway", "n/a"))
+ print(f"Name: {way.tags.get('name', 'n/a')}")
+ print(f" Highway: {way.tags.get('highway', 'n/a')}")
  print(" Nodes:")
  for node in way.nodes:
  print(f" Lat: {node.lat:f}, Lon: {node.lon:f}")

overpy/__about__.py

@@ -19,4 +19,4 @@
 __email__ = ""
 
 __license__ = "MIT"
-__copyright__ = "Copyright 2014-2023 %s" % __author__
+__copyright__ = f"Copyright 2014-2023 {__author__}"

overpy/__init__.py

@@ -1,6 +1,6 @@
-from collections import OrderedDict
 from datetime import datetime
 from decimal import Decimal
+from functools import partial
 from urllib.request import urlopen
 from urllib.error import HTTPError
 from xml.sax import handler, make_parser
@@ -121,43 +121,31 @@ def query(self, query: Union[bytes, str]) -> "Result":
  if not isinstance(query, bytes):
  query = query.encode("utf-8")
 
- retry_num: int = 0
  retry_exceptions: List[exception.OverPyException] = []
- do_retry: bool = True if self.max_retry_count > 0 else False
- while retry_num <= self.max_retry_count:
- if retry_num > 0:
+
+ for run in range(self.max_retry_count + 1):
+ if run:
  time.sleep(self.retry_timeout)
- retry_num += 1
+
+ response = b""
  try:
- f = urlopen(self.url, query)
- except HTTPError as e:
- f = e
-
- response = f.read(self.read_chunk_size)
- while True:
- data = f.read(self.read_chunk_size)
- if len(data) == 0:
- break
- response = response + data
- f.close()
+ 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
+ except HTTPError as exc:
+ f = exc
 
  current_exception: exception.OverPyException
  if f.code == 200:
  content_type = f.getheader("Content-Type")
-
  if content_type == "application/json":
  return self.parse_json(response)
-
- if content_type == "application/osm3s+xml":
+ elif content_type == "application/osm3s+xml":
  return self.parse_xml(response)
-
- current_exception = exception.OverpassUnknownContentType(content_type)
- if not do_retry:
- raise current_exception
- retry_exceptions.append(current_exception)
- continue
-
- if f.code == 400:
+ else:
+ current_exception = exception.OverpassUnknownContentType(content_type)
+ elif f.code == 400:
  msgs: List[str] = []
  for msg_raw in self._regex_extract_error_msg.finditer(response):
  msg_clean_bytes = self._regex_remove_tag.sub(b"", msg_raw.group("msg"))
@@ -166,37 +154,17 @@ def query(self, query: Union[bytes, str]) -> "Result":
  except UnicodeDecodeError:
  msg = repr(msg_clean_bytes)
  msgs.append(msg)
-
- current_exception = exception.OverpassBadRequest(
- query,
- msgs=msgs
- )
- if not do_retry:
- raise current_exception
- retry_exceptions.append(current_exception)
- continue
-
- if f.code == 429:
+ current_exception = exception.OverpassBadRequest(query, msgs=msgs)
+ elif f.code == 429:
  current_exception = exception.OverpassTooManyRequests()
- if not do_retry:
- raise current_exception
- retry_exceptions.append(current_exception)
- continue
-
- if f.code == 504:
+ elif f.code == 504:
  current_exception = exception.OverpassGatewayTimeout()
- if not do_retry:
- raise current_exception
- retry_exceptions.append(current_exception)
- continue
-
- current_exception = exception.OverpassUnknownHTTPStatusCode(f.code)
- if not do_retry:
+ else:
+ current_exception = exception.OverpassUnknownHTTPStatusCode(f.code)
+ if not self.max_retry_count:
  raise current_exception
  retry_exceptions.append(current_exception)
- continue
-
- raise exception.MaxRetriesReached(retry_count=retry_num, exceptions=retry_exceptions)
+ raise exception.MaxRetriesReached(retry_count=run + 1, exceptions=retry_exceptions)
 
  def parse_json(self, data: Union[bytes, str], encoding: str = "utf-8") -> "Result":
  """
@@ -250,18 +218,18 @@ def __init__(
  """
  if elements is None:
  elements = []
- self._areas: Dict[int, Union["Area", "Node", "Relation", "Way"]] = OrderedDict(
- (element.id, element) for element in elements if is_valid_type(element, Area)
- )
- self._nodes = OrderedDict(
- (element.id, element) for element in elements if is_valid_type(element, Node)
- )
- self._ways = OrderedDict(
- (element.id, element) for element in elements if is_valid_type(element, Way)
- )
- self._relations = OrderedDict(
- (element.id, element) for element in elements if is_valid_type(element, Relation)
- )
+ self._areas: Dict[int, Union["Area", "Node", "Relation", "Way"]] = {
+ element.id: element for element in elements if is_valid_type(element, Area)
+ }
+ self._nodes = {
+ element.id: element for element in elements if is_valid_type(element, Node)
+ }
+ self._ways = {
+ element.id: element for element in elements if is_valid_type(element, Way)
+ }
+ self._relations = {
+ element.id: element for element in elements if is_valid_type(element, Relation)
+ }
  self._class_collection_map: Dict[Any, Any] = {
  Node: self._nodes,
  Way: self._ways,
@@ -438,12 +406,9 @@ def get_area(self, area_id: int, resolve_missing: bool = False) -> "Area":
 
  query = ("\n"
  "[out:json];\n"
- "area({area_id});\n"
+ f"area({area_id});\n"
  "out body;\n"
  )
- query = query.format(
- area_id=area_id
- )
  tmp_result = self.api.query(query)
  self.expand(tmp_result)
 
@@ -480,12 +445,9 @@ def get_node(self, node_id: int, resolve_missing: bool = False) -> "Node":
 
  query = ("\n"
  "[out:json];\n"
- "node({node_id});\n"
+ f"node({node_id});\n"
  "out body;\n"
  )
- query = query.format(
- node_id=node_id
- )
  tmp_result = self.api.query(query)
  self.expand(tmp_result)
 
@@ -523,12 +485,9 @@ def get_relation(self, rel_id: int, resolve_missing: bool = False) -> "Relation"
 
  query = ("\n"
  "[out:json];\n"
- "relation({relation_id});\n"
+ f"relation({rel_id});\n"
  "out body;\n"
  )
- query = query.format(
- relation_id=rel_id
- )
  tmp_result = self.api.query(query)
  self.expand(tmp_result)
 
@@ -565,12 +524,9 @@ def get_way(self, way_id: int, resolve_missing: bool = False) -> "Way":
 
  query = ("\n"
  "[out:json];\n"
- "way({way_id});\n"
+ f"way({way_id});\n"
  "out body;\n"
  )
- query = query.format(
- way_id=way_id
- )
  tmp_result = self.api.query(query)
  self.expand(tmp_result)
 
@@ -950,13 +906,10 @@ def get_nodes(self, resolve_missing: bool = False) -> List[Node]:
 
  query = ("\n"
  "[out:json];\n"
- "way({way_id});\n"
+ f"way({self.id});\n"
  "node(w);\n"
  "out body;\n"
  )
- query = query.format(
- way_id=self.id
- )
  tmp_result = self._result.api.query(query)
  self._result.expand(tmp_result)
  resolved = True
@@ -1423,9 +1376,9 @@ def startElement(self, name: str, attrs: dict):
  if name in self.ignore_start:
  return
  try:
- handler = getattr(self, '_handle_start_%s' % name)
+ handler = getattr(self, f"_handle_start_{name}")
  except AttributeError:
- raise KeyError("Unknown element start '%s'" % name)
+ raise KeyError(f"Unknown element start {name!r}")
  handler(attrs)
 
  def endElement(self, name: str):
@@ -1437,9 +1390,9 @@ def endElement(self, name: str):
  if name in self.ignore_end:
  return
  try:
- handler = getattr(self, '_handle_end_%s' % name)
+ handler = getattr(self, f"_handle_end_{name}")
  except AttributeError:
- raise KeyError("Unknown element end '%s'" % name)
+ raise KeyError(f"Unknown element end {name!r}")
  handler()
 
  def _handle_start_center(self, attrs: dict):
@@ -1617,7 +1570,7 @@ def _handle_start_member(self, attrs: dict):
  }
  cls: Type[RelationMember] = cls_map.get(attrs["type"])
  if cls is None:
- raise ValueError("Undefined type for member: '%s'" % attrs['type'])
+ raise ValueError(f"Undefined type for member: {attrs['type']!r}")
 
  self.cur_relation_member = cls(**params)
  self._curr['members'].append(self.cur_relation_member)

overpy/exception.py

@@ -31,10 +31,7 @@ def __init__(self, type_expected, type_provided=None):
  self.type_provided = type_provided
 
  def __str__(self) -> str:
- return "Type expected '{}' but '{}' provided".format(
- self.type_expected,
- str(self.type_provided)
- )
+ return f"Type expected {self.type_expected!r} but {self.type_provided!r} provided"
 
 
 class MaxRetriesReached(OverPyException):
@@ -46,7 +43,7 @@ def __init__(self, retry_count, exceptions):
  self.retry_count = retry_count
 
  def __str__(self) -> str:
- return "Unable get any result from the Overpass API server after %d retries." % self.retry_count
+ return f"Unable get any result from the Overpass API server after {self.retry_count} retries."
 
 
 class OverpassBadRequest(OverPyException):
@@ -142,7 +139,7 @@ def __init__(self, content_type):
  def __str__(self) -> str:
  if self.content_type is None:
  return "No content type returned"
- return "Unknown content type: %s" % self.content_type
+ return f"Unknown content type: {self.content_type}"
 
 
 class OverpassUnknownError(OverpassError):
@@ -163,4 +160,4 @@ def __init__(self, code):
  self.code = code
 
  def __str__(self) -> str:
- return "Unknown/Unhandled status code: %d" % self.code
+ return f"Unknown/Unhandled status code: {self.code}"

overpy/helper.py

@@ -20,10 +20,10 @@ def get_street(
  if api is None:
  api = overpy.Overpass()
 
- query = """
- area(%s)->.location;
+ query = f"""
+ area({areacode})->.location;
  (
- way[highway][name="%s"](area.location);
+ way[highway][name="{street}"](area.location);
  - (
  way[highway=service](area.location);
  way[highway=track](area.location);
@@ -34,7 +34,7 @@ def get_street(
  out skel qt;
  """
 
- data = api.query(query % (areacode, street))
+ data = api.query(query)
 
  return data
 
@@ -57,16 +57,16 @@ def get_intersection(
  if api is None:
  api = overpy.Overpass()
 
- query = """
- area(%s)->.location;
+ query = f"""
+ area({areacode}->.location;
  (
- way[highway][name="%s"](area.location); node(w)->.n1;
- way[highway][name="%s"](area.location); node(w)->.n2;
+ way[highway][name="{street1}"](area.location); node(w)->.n1;
+ way[highway][name="{street2}"](area.location); node(w)->.n2;
  );
  node.n1.n2;
  out meta;
  """
 
- data = api.query(query % (areacode, street1, street2))
+ data = api.query(query)
 
  return data.get_nodes()

setup.cfg

@@ -8,7 +8,6 @@ classifiers =
  Operating System :: OS Independent
  Programming Language :: Python
  Programming Language :: Python :: 3
- Programming Language :: Python :: 3.6
  Programming Language :: Python :: 3.7
  Programming Language :: Python :: 3.8
  Programming Language :: Python :: 3.9
@@ -20,7 +19,7 @@ project_urls =
  Issue Tracker = https://github.com/DinoTools/python-overpy/issues
 
 [options]
-python_requires = >=3.6
+python_requires = >=3.7
 include_package_data = true
 zip_safe = false