Extend routing API to fetch and combine route-specific preferences with global preferences

app/api/v1/endpoints/routes.py

@@ -19,6 +19,7 @@
 from app.services.ai_agents_service import ai_agents_service
 from app.services.auth_service import auth_service
 from app.services.global_preference_service import global_preference_service
+from app.services.route_preference_service import route_preference_service
 from app.services.routing_service import (
     RoutingAPIError,
     RoutingDataError,
@@ -78,13 +79,28 @@ async def search_routes(
 
     # 2. Add stored global preferences from authenticated user
     try:
-        stored_prefs = global_preference_service.get_user_preferences(
+        global_prefs = global_preference_service.get_user_preferences(
             db, int(current_user.id)
         )
-        for pref in stored_prefs:
+        for pref in global_prefs:
             user_preferences.append({"prompt": pref.prompt})
     except Exception as e:  # pylint: disable=broad-except
-        logger.warning("Failed to fetch user preferences: %s", str(e))
+        logger.warning("Failed to fetch global preferences: %s", str(e))
+
+    # 3. Add route-specific preferences that match the coordinates
+    try:
+        route_prefs = route_preference_service.get_preferences_by_coordinates(
+            db,
+            int(current_user.id),
+            request.origin.latitude,
+            request.origin.longitude,
+            request.destination.latitude,
+            request.destination.longitude,
+        )
+        for pref in route_prefs:
+            user_preferences.append({"prompt": pref.prompt})
+    except Exception as e:  # pylint: disable=broad-except
+        logger.warning("Failed to fetch route-specific preferences: %s", str(e))
 
     logger.info(
         "Using %d user preferences for route insights", len(user_preferences)

app/services/route_preference_service.py

@@ -34,6 +34,41 @@ def get_user_preferences(
             .all()
         )
 
+    @staticmethod
+    def get_preferences_by_coordinates(
+        db: Session,
+        user_id: int,
+        from_latitude: float,
+        from_longitude: float,
+        to_latitude: float,
+        to_longitude: float,
+    ) -> List[RoutePreference]:
+        """
+        Get route preferences for a user that match specific coordinates.
+
+        Args:
+            db: Database session
+            user_id: User ID to get preferences for
+            from_latitude: Origin latitude
+            from_longitude: Origin longitude
+            to_latitude: Destination latitude
+            to_longitude: Destination longitude
+
+        Returns:
+            List of RoutePreference objects matching the coordinates
+        """
+        return (
+            db.query(RoutePreference)
+            .filter(
+                RoutePreference.user_id == user_id,
+                RoutePreference.from_latitude == from_latitude,
+                RoutePreference.from_longitude == from_longitude,
+                RoutePreference.to_latitude == to_latitude,
+                RoutePreference.to_longitude == to_longitude,
+            )
+            .all()
+        )
+
     @staticmethod
     def get_preference_by_id(
         db: Session, preference_id: int