3pp-mirror/map-machine
1
0
Fork 0
mirror of https://github.com/enzet/map-machine.git synced 2025-05-12 16:46:50 +02:00
Code Issues Projects Releases Packages Wiki Activity

Add error handling.

Browse source
This commit is contained in:
Sergey Vartanov 2021-08-06 00:33:18 +03:00
parent 088fc58870
commit d321f75e21
1 changed files with 12 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

13
roentgen/osm_getter.py
View file

@ -6,6 +6,7 @@ import urllib
from pathlib import Path from pathlib import Path
from typing import Dict, Optional from typing import Dict, Optional
import logging
import urllib3 import urllib3
__author__ = "Sergey Vartanov" __author__ = "Sergey Vartanov"
@ -29,11 +30,13 @@ def get_osm(
if not to_update and result_file_name.is_file(): if not to_update and result_file_name.is_file():
return result_file_name.open().read() return result_file_name.open().read()
content = get_data( content: Optional[bytes] = get_data(
"api.openstreetmap.org/api/0.6/map", "api.openstreetmap.org/api/0.6/map",
{"bbox": boundary_box}, {"bbox": boundary_box},
is_secure=True, is_secure=True,
) )
if content is None:
return None
if BoundaryBox.from_text(boundary_box) is None: if BoundaryBox.from_text(boundary_box) is None:
return None return None
@ -44,7 +47,7 @@ def get_osm(
def get_data( def get_data(
address: str, parameters: Dict[str, str], is_secure: bool = False address: str, parameters: Dict[str, str], is_secure: bool = False
) -> bytes: ) -> Optional[bytes]:
""" """
Construct Internet page URL and get its descriptor. Construct Internet page URL and get its descriptor.
@ -59,7 +62,13 @@ def get_data(
print("Getting " + url + "...") print("Getting " + url + "...")
pool_manager = urllib3.PoolManager() pool_manager = urllib3.PoolManager()
urllib3.disable_warnings() urllib3.disable_warnings()
try:
result = pool_manager.request("GET", url) result = pool_manager.request("GET", url)
except urllib3.exceptions.MaxRetryError:
logging.fatal("Too many attempts.")
return None
pool_manager.clear() pool_manager.clear()
time.sleep(2) time.sleep(2)
return result.data return result.data