cleanup
Some checks failed
Build and deploy the backend to staging / Build and push image (pull_request) Successful in 2m16s
Run linting on the backend code / Build (pull_request) Successful in 28s
Run testing on the backend code / Build (pull_request) Failing after 1m14s
Build and deploy the backend to staging / Deploy to staging (pull_request) Successful in 24s
Some checks failed
Build and deploy the backend to staging / Build and push image (pull_request) Successful in 2m16s
Run linting on the backend code / Build (pull_request) Successful in 28s
Run testing on the backend code / Build (pull_request) Failing after 1m14s
Build and deploy the backend to staging / Deploy to staging (pull_request) Successful in 24s
This commit is contained in:
@@ -1,12 +1,13 @@
|
||||
"""Module allowing connexion to overpass api and fectch data from OSM."""
|
||||
from typing import Literal, List
|
||||
import urllib
|
||||
import json
|
||||
import logging
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
from .caching_strategy import get_cache_key, CachingStrategy
|
||||
|
||||
|
||||
logger = logging.getLogger('overpass')
|
||||
ElementTypes = List[Literal['way', 'node', 'relation']]
|
||||
|
||||
|
||||
@@ -46,8 +47,13 @@ def build_query(area: tuple, element_types: ElementTypes,
|
||||
query = '('
|
||||
|
||||
# Round the radius to nearest 50 and coordinates to generate less queries
|
||||
search_radius = round(area[0] / 50) * 50
|
||||
loc = tuple((round(area[1], 2), round(area[2], 2)))
|
||||
if area[0] > 500 :
|
||||
search_radius = round(area[0] / 50) * 50
|
||||
loc = tuple((round(area[1], 2), round(area[2], 2)))
|
||||
else :
|
||||
search_radius = round(area[0] / 25) * 25
|
||||
loc = tuple((round(area[1], 3), round(area[2], 3)))
|
||||
|
||||
search_area = f"(around:{search_radius}, {str(loc[0])}, {str(loc[1])})"
|
||||
|
||||
if conditions :
|
||||
@@ -80,7 +86,7 @@ def send_query(query: str) -> dict:
|
||||
# Try to fetch the result from the cache
|
||||
cached_response = CachingStrategy.get(cache_key)
|
||||
if cached_response is not None :
|
||||
print("Cache hit!")
|
||||
logger.debug("Cache hit.")
|
||||
return cached_response
|
||||
|
||||
# Define the Overpass API endpoint
|
||||
@@ -106,12 +112,9 @@ def send_query(query: str) -> dict:
|
||||
|
||||
# Cache the response data as an ElementTree root
|
||||
CachingStrategy.set(cache_key, root)
|
||||
logger.debug("Response data added to cache.")
|
||||
|
||||
return root
|
||||
|
||||
except urllib.error.URLError as e:
|
||||
print(f"Error connecting to Overpass API: {e}")
|
||||
return None
|
||||
except json.JSONDecodeError:
|
||||
print("Error decoding the JSON response from Overpass API.")
|
||||
return None
|
||||
raise ConnectionError(f"Error connecting to Overpass API: {e}") from e
|
||||
|
Reference in New Issue
Block a user