Fetch data from OSM Nominatim

app/crud.py

@@ -189,9 +189,19 @@ def create_location(db: Session, location: LocationCreate):
 
 
 def get_or_create_location(db: Session, location: LocationCreate):
+    created = False
     db_location = get_location_by_osm_id_and_type(
         db, osm_id=location.osm_id, osm_type=location.osm_type
     )
     if not db_location:
         db_location = create_location(db, location=location)
-    return db_location
+        created = True
+    return db_location, created
+
+
+def update_location(db: Session, location: LocationBase, update_dict: dict):
+    for key, value in update_dict.items():
+        setattr(location, key, value)
+    db.commit()
+    db.refresh(location)
+    return location

app/tasks.py

@@ -2,6 +2,7 @@
 
 from app import crud
 from app.schemas import LocationCreate, PriceBase
+from app.utils import fetch_location_openstreetmap_details
 
 
 def create_price_location(db: Session, price: PriceBase):
@@ -10,6 +11,15 @@ def create_price_location(db: Session, price: PriceBase):
         location = LocationCreate(
             osm_id=price.location_osm_id, osm_type=price.location_osm_type
         )
-        db_location = crud.get_or_create_location(db, location=location)
+        db_location, created = crud.get_or_create_location(db, location=location)
         # link the location to the price
         crud.set_price_location(db, price=price, location=db_location)
+        # fetch data from OpenStreetMap if created
+        if created:
+            location_openstreetmap_details = fetch_location_openstreetmap_details(
+                location=db_location
+            )
+            if location_openstreetmap_details:
+                crud.update_location(
+                    db, location=db_location, update_dict=location_openstreetmap_details
+                )

app/utils.py

@@ -1,9 +1,12 @@
 import logging
 
 import sentry_sdk
+from OSMPythonTools.nominatim import Nominatim
 from sentry_sdk.integrations import Integration
 from sentry_sdk.integrations.logging import LoggingIntegration
 
+from app.schemas import LocationBase
+
 
 def init_sentry(sentry_dsn: str | None, integrations: list[Integration] | None = None):
     if sentry_dsn:
@@ -18,3 +21,31 @@ def init_sentry(sentry_dsn: str | None, integrations: list[Integration] | None =
             sentry_dsn,
             integrations=integrations,
         )
+
+
+def openstreetmap_nominatim_search(osm_id: int, osm_type: str):
+    client = Nominatim()
+    search_query = f"{osm_type}/{osm_id}"
+    return client.query(search_query, lookup=True).toJSON()
+
+
+def fetch_location_openstreetmap_details(location: LocationBase):
+    location_openstreetmap_details = dict()
+    try:
+        response = openstreetmap_nominatim_search(
+            osm_id=location.osm_id, osm_type=location.osm_type.value.lower()
+        )
+        if len(response):
+            if "display_name" in response[0]:
+                location_openstreetmap_details["osm_display_name"] = response[0][
+                    "display_name"
+                ]
+            if "lat" in response[0]:
+                location_openstreetmap_details["osm_lat"] = response[0]["lat"]
+            if "lon" in response[0]:
+                location_openstreetmap_details["osm_lon"] = response[0]["lon"]
+        return location_openstreetmap_details
+    except Exception as e:
+        print("fetch_location_openstreetmap_details: error returned from OpenStreetMap")
+        print(e)
+        return