fix: operations api async calls

.github/workflows/build-test.yml

@@ -287,4 +287,93 @@ jobs:
         with:
           name: coverage_report_${{ matrix.group }}
           path: scripts/coverage_reports/
-          overwrite: true
+          overwrite: true
+
+  operations-api-surface:
+    needs: setup
+    if: ${{ needs.setup.outputs.should-run-tests == 'true' }}
+    runs-on: ubuntu-latest
+    name: Operations API surface
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v7
+
+      - uses: actions/setup-python@v7
+        with:
+          python-version: ${{ env.python_version }}
+
+      - name: Download Operations API generated code
+        uses: actions/download-artifact@v8
+        with:
+          name: feeds_operations_gen
+          path: functions-python/operations_api/src/feeds_gen/
+
+      # The app imports the generated SQLAlchemy models via the shared symlinks
+      # (shared.database_gen / shared.users_database_gen), so those artifacts must
+      # be present for the server to start.
+      - name: Download DB models
+        uses: actions/download-artifact@v8
+        with:
+          name: database_gen
+          path: api/src/shared/database_gen/
+
+      - name: Download users DB models
+        uses: actions/download-artifact@v8
+        with:
+          name: users_database_gen
+          path: api/src/shared/users_database_gen/
+
+      - name: Docker Compose DB
+        run: docker compose --env-file ./config/.env.local up -d postgres-test --wait
+        working-directory: ${{ github.workspace }}
+
+      - name: Set up JDK ${{ env.java_version }}
+        uses: actions/setup-java@v5
+        with:
+          java-version: ${{ env.java_version }}
+          distribution: 'temurin'
+
+      - name: Cache Liquibase
+        id: cache-liquibase
+        uses: actions/cache@v6
+        with:
+          path: ~/.liquibase
+          key: liquibase-${{ env.liquibase_version }}
+
+      - name: Install Liquibase
+        if: steps.cache-liquibase.outputs.cache-hit != 'true'
+        env:
+          LIQUIBASE_VERSION: ${{ env.liquibase_version }}
+        run: |
+          curl -sSL https://github.com/liquibase/liquibase/releases/download/v${LIQUIBASE_VERSION}/liquibase-${LIQUIBASE_VERSION}.tar.gz -o liquibase.tar.gz
+          rm -rf liquibase-dist
+          mkdir liquibase-dist
+          tar -xzf liquibase.tar.gz -C liquibase-dist
+          mv liquibase-dist ~/.liquibase
+
+      - name: Link Liquibase
+        run: sudo ln -sf ~/.liquibase/liquibase /usr/local/bin/liquibase
+
+      - name: Run Liquibase on test DB
+        working-directory: ${{ github.workspace }}/liquibase
+        run: |
+          export LIQUIBASE_COMMAND_CHANGELOG_FILE="changelog.xml"
+          export LIQUIBASE_COMMAND_URL=jdbc:postgresql://localhost:54320/MobilityDatabaseTest
+          export LIQUIBASE_COMMAND_USERNAME=postgres
+          export LIQUIBASE_COMMAND_PASSWORD=postgres
+          liquibase update
+
+      - name: Run Liquibase on users test DB
+        working-directory: ${{ github.workspace }}/liquibase
+        run: |
+          export LIQUIBASE_COMMAND_CHANGELOG_FILE="changelog_user.xml"
+          export LIQUIBASE_COMMAND_URL=jdbc:postgresql://localhost:54320/${POSTGRES_USER_TEST_DB:-MobilityDatabaseUsersTest}
+          export LIQUIBASE_COMMAND_USERNAME=postgres
+          export LIQUIBASE_COMMAND_PASSWORD=postgres
+          liquibase update
+
+      # DB is already up + migrated by the steps above, so run with --no-docker.
+      # The script provisions the venv/deps/symlinks, starts the API, and hits every endpoint.
+      - name: Run Operations API surface suite
+        shell: bash
+        run: scripts/api-operations-surface.sh --no-docker

functions-python/helpers/pub_sub.py

@@ -77,7 +77,18 @@ def trigger_dataset_download(
     execution_id: str,
     topic_name: str = DATASET_BATCH_TOPIC,
 ) -> None:
-    """Publishes the feed to the configured Pub/Sub topic."""
+    """Publishes the feed to the configured Pub/Sub topic.
+
+    No-ops when no topic is configured (``DATASET_PROCESSING_TOPIC_NAME`` unset),
+    e.g. local development and tests, so the caller never constructs a Pub/Sub
+    client or blocks on a publish. Mirrors ``create_web_revalidation_task``.
+    """
+    if not topic_name:
+        logging.info(
+            "DATASET_PROCESSING_TOPIC_NAME not set; skipping dataset download for feed %s",
+            getattr(feed, "stable_id", None),
+        )
+        return
     publisher = get_pubsub_client()
     topic_path = publisher.topic_path(PROJECT_ID, topic_name)
     logging.info("Publishing to Pub/Sub topic: %s", topic_path)

functions-python/operations_api/src/feeds_operations/impl/feeds_operations_impl.py

@@ -19,7 +19,7 @@
 
 import logging
 from datetime import datetime
-from typing import Annotated, Optional
+from typing import Annotated, Final, Optional
 
 from deepdiff import DeepDiff
 from fastapi import HTTPException
@@ -89,6 +89,38 @@
 from .models.operation_gtfs_rt_feed_impl import OperationGtfsRtFeedImpl
 from .request_validator import validate_request
 
+# Fields projected by `from_orm` (see UpdateRequestGtfs(Rt)FeedImpl) that are derived
+# from related rows and never written back by `to_orm`, so an update request cannot set
+# them. `license_is_spdx` is derived from the License relationship (feed.license.is_spdx);
+# comparing it would report a phantom change whenever the request omits it.
+_DERIVED_SOURCE_INFO_FIELDS: Final[tuple[str, ...]] = ("license_is_spdx",)
+
+
+def _normalize_for_diff(value):
+    """Recursively coerce "absent" representations to None so change detection mirrors
+    the write path (`to_orm`), which persists empty/falsy values as None. Applied to both
+    sides of the diff for every field: an empty string or empty list is treated as "no
+    value" rather than a change, so None-vs-"" and None-vs-[] never register as edits.
+    """
+    if isinstance(value, dict):
+        return {key: _normalize_for_diff(item) for key, item in value.items()}
+    if isinstance(value, list):
+        normalized = [_normalize_for_diff(item) for item in value]
+        return None if not normalized else normalized
+    if value == "":
+        return None
+    return value
+
+
+def _strip_derived_fields(dumped: dict) -> dict:
+    """Drop read-only/derived fields from a dumped update model so they do not register
+    as changes (they are not settable through the update request)."""
+    source_info = dumped.get("source_info")
+    if isinstance(source_info, dict):
+        for field in _DERIVED_SOURCE_INFO_FIELDS:
+            source_info.pop(field, None)
+    return dumped
+
 
 class OperationsApiImpl(BaseOperationsApi):
     """Implementation of the operations API."""
@@ -282,9 +314,15 @@ def detect_changes(
         copy_feed.operational_status_action = (
             update_request_feed.operational_status_action
         )
+        current_values = _strip_derived_fields(
+            _normalize_for_diff(copy_feed.model_dump())
+        )
+        requested_values = _strip_derived_fields(
+            _normalize_for_diff(update_request_feed.model_dump())
+        )
         diff = DeepDiff(
-            copy_feed.model_dump(),
-            update_request_feed.model_dump(),
+            current_values,
+            requested_values,
             ignore_order=True,
         )
         if diff.affected_paths:
@@ -296,7 +334,7 @@ def detect_changes(
         return diff
 
     @validate_request(UpdateRequestGtfsFeed, "update_request_gtfs_feed")
-    async def update_gtfs_feed(
+    def update_gtfs_feed(
         self,
         update_request_gtfs_feed: Annotated[
             UpdateRequestGtfsFeed,
@@ -310,10 +348,10 @@ async def update_gtfs_feed(
             - 400: Feed ID not found.
             - 500: Internal server error.
         """
-        return await self._update_feed(update_request_gtfs_feed, DataType.GTFS)
+        return self._update_feed(update_request_gtfs_feed, DataType.GTFS)
 
     @validate_request(UpdateRequestGtfsRtFeed, "update_request_gtfs_rt_feed")
-    async def update_gtfs_rt_feed(
+    def update_gtfs_rt_feed(
         self,
         update_request_gtfs_rt_feed: Annotated[
             UpdateRequestGtfsRtFeed,
@@ -327,10 +365,10 @@ async def update_gtfs_rt_feed(
             - 400: Feed ID not found.
             - 500: Internal server error.
         """
-        return await self._update_feed(update_request_gtfs_rt_feed, DataType.GTFS_RT)
+        return self._update_feed(update_request_gtfs_rt_feed, DataType.GTFS_RT)
 
     @with_db_session
-    async def _update_feed(
+    def _update_feed(
         self,
         update_request_feed: UpdateRequestGtfsFeed | UpdateRequestGtfsRtFeed,
         data_type: DataType,
@@ -340,7 +378,7 @@ async def _update_feed(
         Update the specified feed in the Mobility Database
         """
         try:
-            feed_from_db = await OperationsApiImpl.fetch_feed(
+            feed_from_db = OperationsApiImpl.fetch_feed(
                 data_type, db_session, update_request_feed
             )
 
@@ -365,7 +403,7 @@ async def _update_feed(
                     r.target_id for r in getattr(feed_from_db, "redirectingids", [])
                 }
 
-                await OperationsApiImpl._populate_feed_values(
+                OperationsApiImpl._populate_feed_values(
                     feed_from_db, impl_class, db_session, update_request_feed
                 )
                 if getattr(update_request_feed, "propagate_license", False):
@@ -459,7 +497,7 @@ async def _update_feed(
             raise HTTPException(status_code=500, detail=f"Internal server error: {e}")
 
     @staticmethod
-    async def _populate_feed_values(feed, impl_class, session, update_request_feed):
+    def _populate_feed_values(feed, impl_class, session, update_request_feed):
         impl_class.to_orm(update_request_feed, feed, session)
         action = update_request_feed.operational_status_action
         # This is a temporary solution as the operational_status is not visible in the diff
@@ -473,7 +511,7 @@ async def _populate_feed_values(feed, impl_class, session, update_request_feed):
         session.add(feed)
 
     @staticmethod
-    async def fetch_feed(data_type, session, update_request_feed):
+    def fetch_feed(data_type, session, update_request_feed):
         """Fetch a feed by its stable ID with eager loading.
 
         Args:
@@ -499,7 +537,7 @@ async def fetch_feed(data_type, session, update_request_feed):
         return feed
 
     @with_db_session
-    async def create_gtfs_feed(
+    def create_gtfs_feed(
         self,
         operation_create_request_gtfs_feed: Annotated[
             OperationCreateRequestGtfsFeed,
@@ -559,7 +597,7 @@ async def create_gtfs_feed(
         return JSONResponse(status_code=201, content=jsonable_encoder(payload))
 
     @with_db_session
-    async def create_gtfs_rt_feed(
+    def create_gtfs_rt_feed(
         self,
         operation_create_request_gtfs_rt_feed: Annotated[
             OperationCreateRequestGtfsRtFeed,

functions-python/operations_api/src/feeds_operations/impl/licenses_api_impl.py

@@ -68,7 +68,7 @@ def handle_get_license(
         # Build Pydantic model from ORM object attributes
         return LicenseWithRulesImpl.from_orm(license_orm)
 
-    async def get_license(
+    def get_license(
         self,
         id: StrictStr,
     ) -> LicenseWithRules:
@@ -136,7 +136,7 @@ def handle_get_licenses(
 
         return [LicenseBaseImpl.from_orm(item) for item in items]
 
-    async def get_licenses(
+    def get_licenses(
         self,
         offset: str = "0",
         limit: str = "100",
@@ -173,7 +173,7 @@ def handle_get_matching_licenses(
                 status_code=500, detail="Error retrieving matching licenses"
             )
 
-    async def get_matching_licenses(
+    def get_matching_licenses(
         self,
         get_matching_licenses_request: GetMatchingLicensesRequest,
     ) -> List[MatchingLicense]:
@@ -222,7 +222,7 @@ def handle_propagate_match_license(
             logging.error("Error in propagate_match_license: %s", e)
             raise HTTPException(status_code=500, detail="Internal server error")
 
-    async def propagate_match_license(
+    def propagate_match_license(
         self,
         propagate_license_request: PropagateLicenseRequest,
     ) -> PropagateLicenseResponse:

functions-python/operations_api/src/feeds_operations/impl/request_validator.py

@@ -29,7 +29,7 @@ def validate_request(model: BaseModel, parameter_name: str, validate_none: bool
 
     def decorator(func):
         @wraps(func)
-        async def wrapper(*args, **kwargs):
+        def wrapper(*args, **kwargs):
             func_args = inspect.getfullargspec(func).args
             print(func_args)
             index = func_args.index(parameter_name)
@@ -42,7 +42,7 @@ async def wrapper(*args, **kwargs):
             else:
                 if validate_none:
                     raise HTTPException(status_code=400, detail="Missing parameter")
-            return await func(*args, **kwargs)
+            return func(*args, **kwargs)
 
         return wrapper