added core geotagging logic & api

core.py

@@ -0,0 +1,94 @@
+"""Core logic for fetching geotags from addresses or coordinates."""
+
+import re
+from typing import List
+import httpx
+import h3
+from .models import GeotagResult
+
+# Regex for validating "latitude,longitude" input
+COORD_REGEX = re.compile(r"^-?([1-8]?[0-9]|[1-9]0)\.{1}\d{1,15}\s*,\s*-?([1]?[0-7]?[0-9]|[1]?[1-8]?[0])\.{1}\d{1,15}$")
+
+class GeotaggingError(Exception):
+    """Base exception for geotagging errors."""
+    pass
+
+class InvalidCoordinatesError(GeotaggingError):
+    """Raised when coordinate input is invalid."""
+    pass
+
+class GeocodingServiceError(GeotaggingError):
+    """Raised when the external geocoding service fails."""
+    pass
+
+class NoResultsFoundError(GeotaggingError):
+    """Raised when no results are found for a query."""
+    pass
+
+
+async def fetch_geotags(q: str, resolution: int) -> List[GeotagResult]:
+    """
+    Fetches geotag information for a given query string (address or lat,lng).
+
+    Args:
+        q: The address string or 'latitude,longitude' pair.
+        resolution: The H3 resolution (0-15).
+
+    Returns:
+        A list of GeotagResult objects.
+
+    Raises:
+        InvalidCoordinatesError: If the input coordinates are malformed.
+        GeocodingServiceError: If there's an issue connecting to the geocoding service.
+        NoResultsFoundError: If the geocoding service returns no results.
+    """
+    locations = []
+
+    if COORD_REGEX.match(q.replace(" ", "")):
+        try:
+            lat_str, lon_str = q.split(',')
+            lat, lon = float(lat_str), float(lon_str)
+            locations.append({
+                "lat": lat,
+                "lon": lon,
+                "display_name": f"Coordinates: {lat:.6f}, {lon:.6f}"
+            })
+        except (ValueError, IndexError):
+            raise InvalidCoordinatesError("Invalid coordinate format. Use 'latitude,longitude'.")
+    else:
+        headers = {'User-Agent': 'TrackEmDown/1.0 (nikhilsingh.io)'}
+        url = f"https://nominatim.openstreetmap.org/search?format=json&q={q}"
+
+        async with httpx.AsyncClient() as client:
+            try:
+                response = await client.get(url, headers=headers, timeout=10.0)
+                response.raise_for_status()
+                locations = response.json()
+            except httpx.RequestError as exc:
+                raise GeocodingServiceError(f"Error connecting to geocoding service: {exc}")
+
+    if not locations:
+        raise NoResultsFoundError("No results found for the given query.")
+
+    results = []
+    for loc in locations:
+        try:
+            lat, lon = float(loc['lat']), float(loc['lon'])
+            geotag = h3.latlng_to_cell(lat, lon, resolution)
+            results.append(
+                GeotagResult(
+                    address=loc.get('display_name', 'N/A'),
+                    latitude=lat,
+                    longitude=lon,
+                    geotag=geotag
+                )
+            )
+        except (ValueError, KeyError) as e:
+            # Skip malformed results from the external API
+            print(f"Skipping a location due to parsing error: {e}")
+            continue
+    
+    if not results:
+        raise NoResultsFoundError("Geocoding service returned malformed data.")
+
+    return results

main.py

@@ -0,0 +1,50 @@
+"""FastAPI application for the Geotagging Service."""
+
+from fastapi import FastAPI, HTTPException, Query
+from .models import GeotagResponse
+from .core import fetch_geotags, NoResultsFoundError, GeocodingServiceError, InvalidCoordinatesError
+
+app = FastAPI(
+    title="Geotagging Service",
+    description="An API to convert addresses and coordinates to Uber H3 geotags using OSM Nominatim.",
+    version="1.0.0",
+)
+
+@app.get(
+    "/geotag",
+    response_model=GeotagResponse,
+    summary="Get H3 geotag for an address or coordinates",
+    tags=["Geotagging"],
+)
+async def get_geotag(
+    q: str = Query(..., description="The address string or 'latitude,longitude' pair to look up."),
+    resolution: int = Query(12, ge=0, le=15, description="The H3 resolution (0-15). Default is 12.")
+):
+    """
+    This endpoint takes a query string `q` and an H3 `resolution` and returns a list of matching geotags.
+
+    - If `q` is a valid "latitude,longitude" string, it's converted directly.
+    - If `q` is an address, it's sent to the OpenStreetMap Nominatim API for geocoding.
+    - The resulting coordinates are then converted into H3 cell identifiers (geotags).
+    """
+    try:
+        results = await fetch_geotags(q, resolution)
+        return GeotagResponse(
+            query=q,
+            resolution=resolution,
+            results=results,
+        )
+    except InvalidCoordinatesError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except NoResultsFoundError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+    except GeocodingServiceError as e:
+        raise HTTPException(status_code=503, detail=str(e))
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"An unexpected error occurred: {e}")
+
+
+@app.get("/", include_in_schema=False)
+async def root():
+    """Root endpoint for health check."""
+    return {"status": "ok", "message": "Geotagging Service is running."}