feat: fetch location data from OpenStreetMap Nominatim

app/api.py

@@ -117,12 +117,12 @@ def authentication(
     data = {"user_id": form_data.username, "password": form_data.password}
     r = requests.post(settings.oauth2_server_url, data=data)  # type: ignore
     if r.status_code == 200:
-        token = await create_token(form_data.username)
+        token = create_token(form_data.username)
         user = schemas.UserBase(user_id=form_data.username, token=token)
         crud.create_user(db, user=user)
         return {"access_token": token, "token_type": "bearer"}
     elif r.status_code == 403:
-        await asyncio.sleep(2)  # prevents brute-force
+        asyncio.sleep(2)  # prevents brute-force
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="Invalid authentication credentials",

app/crud.py

@@ -223,9 +223,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, ProductCreate
+from app.utils import fetch_location_openstreetmap_details
 
 
 def create_price_product(db: Session, price: PriceBase):
@@ -19,6 +20,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,15 @@
 import logging
 
 import sentry_sdk
+from openfoodfacts.utils import get_logger
+from OSMPythonTools.nominatim import Nominatim
 from sentry_sdk.integrations import Integration
 from sentry_sdk.integrations.logging import LoggingIntegration
 
+from app.schemas import LocationBase
+
+logger = get_logger(__name__)
+
 
 def init_sentry(sentry_dsn: str | None, integrations: list[Integration] | None = None):
     if sentry_dsn:
@@ -18,3 +24,34 @@ 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):
+            for osm_field in ["name", "display_name", "lat", "lon"]:
+                if osm_field in response[0]:
+                    location_openstreetmap_details[f"osm_{osm_field}"] = response[0][
+                        osm_field
+                    ]
+            if "address" in response[0]:
+                for osm_address_field in ["postcode", "city", "country"]:
+                    if osm_address_field in response[0]["address"]:
+                        location_openstreetmap_details[
+                            f"osm_address_{osm_address_field}"
+                        ] = response[0]["address"][osm_address_field]
+
+        return location_openstreetmap_details
+    except Exception:
+        logger.exception("error returned from OpenStreetMap")
+        return