feat: ✨ zipcode resolver

Author: marcvergees

app/api/router.py

@@ -5,9 +5,10 @@
 
 from fastapi import APIRouter
 
-from app.api.routes import forms, templates, weather
+from app.api.routes import forms, templates, weather, zipcode
 
 api_router = APIRouter()
 api_router.include_router(templates.router)
 api_router.include_router(forms.router)
-api_router.include_router(weather.router)
+api_router.include_router(weather.router)
+api_router.include_router(zipcode.router)

app/api/routes/zipcode.py

@@ -0,0 +1,42 @@
+from fastapi import APIRouter
+
+from app.core.errors.base import AppError
+from app.services.controller import Controller
+
+router = APIRouter(prefix="/zipcode", tags=["zipcode"])
+
+@router.get("/postal-code")
+def get_postal_code(
+    country: str,
+    postal_code: str,
+):
+    """
+    Fetch data for the given postal code.
+
+    Query params:
+      - country: required string
+      - postal_code: required string
+    """
+    controller = Controller()
+    try:
+        return controller.get_postal_code(country, postal_code)
+    except Exception as e:
+        raise AppError(str(e), status_code=500)
+
+@router.get("/location")
+def get_location(
+    country: str,
+    city: str,
+):
+    """
+    Fetch data for the given location.
+
+    Query params:
+      - country: required string
+      - city: required string
+    """
+    controller = Controller()
+    try:
+        return controller.get_location(country, city)
+    except Exception as e:
+        raise AppError(str(e), status_code=500)

app/services/controller.py

@@ -14,4 +14,10 @@ def prepare_fillable(self, pdf_path: str):
         return self.file_manipulator.prepare_fillable(pdf_path)
 
     def get_weather(self, latitude: float, longitude: float, start_date: str, end_date: str, hourly_fields: list = None):
-        return self.external_apis_coordinator.get_weather(latitude, longitude, start_date, end_date, hourly_fields)
+        return self.external_apis_coordinator.get_weather(latitude, longitude, start_date, end_date, hourly_fields)
+
+    def get_postal_code(self, country: str, postal_code: str):
+        return self.external_apis_coordinator.get_postal_code(country, postal_code)
+        
+    def get_location(self, country: str, city: str):
+        return self.external_apis_coordinator.get_location(country, city)

app/services/external_apis/zipcode_api.py

@@ -0,0 +1,23 @@
+import pgeocode
+import numpy as np
+import pandas as pd
+
+class ZipCodeAPI:
+    def __init__(self):
+        pass
+
+    def get_postal_code(self, country: str, postal_code: str):
+        res = pgeocode.Nominatim(country).query_postal_code(postal_code)
+        if isinstance(res, pd.Series):
+            res = res.replace({np.nan: None}).to_dict()
+        elif isinstance(res, pd.DataFrame):
+            res = res.replace({np.nan: None}).to_dict(orient='records')
+        return res
+
+    def get_location(self, country: str, city: str):
+        res = pgeocode.Nominatim(country).query_location(city)
+        if isinstance(res, pd.Series):
+            res = res.replace({np.nan: None}).to_dict()
+        elif isinstance(res, pd.DataFrame):
+            res = res.replace({np.nan: None}).to_dict(orient='records')
+        return res

app/services/external_apis_coordinator.py

@@ -1,8 +1,16 @@
+from app.services.external_apis.zipcode_api import ZipCodeAPI
 from app.services.external_apis.weather_api import WeatherAPI
 
 class ExternalAPIsCoordinator:
     def __init__(self):
         self.weather_api = WeatherAPI()
+        self.zipcode_api = ZipCodeAPI()
 
     def get_weather(self, latitude: float, longitude: float, start_date: str, end_date: str, hourly_fields: list[str] | None = None):
         return self.weather_api.get_weather(latitude, longitude, start_date, end_date, hourly_fields)
+
+    def get_postal_code(self, country: str, postal_code: str):
+        return self.zipcode_api.get_postal_code(country, postal_code)
+        
+    def get_location(self, country: str, city: str):
+        return self.zipcode_api.get_location(country, city)

examples/pgeocode_demo.py

@@ -0,0 +1,5 @@
+import pgeocode
+
+nomi = pgeocode.Nominatim('us')
+response_pgeo = nomi.query_location("Santa Cruz", top_k=3)
+print(response_pgeo)

requirements.txt

@@ -15,4 +15,5 @@ python-multipart
 openmeteo-requests
 requests-cache
 retry-requests
-pandas
+pandas
+pgeocode