feat: Add create_gtfs_rt_feed endpoint to operations API (#1435)

Author: davidgamez

api/src/shared/db_models/gtfs_rt_feed_impl.py

@@ -1,5 +1,10 @@
+import logging
+
+from sqlalchemy.orm import Session
+
+from shared.database.database import with_db_session
 from shared.db_models.feed_impl import FeedImpl
-from shared.database_gen.sqlacodegen_models import Gtfsrealtimefeed as GtfsRTFeedOrm
+from shared.database_gen.sqlacodegen_models import Gtfsrealtimefeed, Feed as FeedOrm, Entitytype
 from shared.db_models.location_impl import LocationImpl
 from feeds_gen.models.gtfs_rt_feed import GtfsRTFeed
 
@@ -14,11 +19,42 @@ class Config:
         from_attributes = True
 
     @classmethod
-    def from_orm(cls, feed: GtfsRTFeedOrm | None) -> GtfsRTFeed | None:
+    def from_orm(cls, feed: Gtfsrealtimefeed | None) -> GtfsRTFeed | None:
         gtfs_rt_feed: GtfsRTFeed = super().from_orm(feed)
         if not gtfs_rt_feed:
             return None
         gtfs_rt_feed.locations = [LocationImpl.from_orm(item) for item in feed.locations] if feed.locations else []
         gtfs_rt_feed.entity_types = [item.name for item in feed.entitytypes] if feed.entitytypes else []
         gtfs_rt_feed.feed_references = [item.stable_id for item in feed.gtfs_feeds] if feed.gtfs_feeds else []
         return gtfs_rt_feed
+
+    @classmethod
+    @with_db_session
+    def to_orm_entity_types(cls, entity_types: list, db_session: Session) -> list[Entitytype]:
+        """Convert the entity_types list to a list of Entitytype ORM objects."""
+        if not entity_types:
+            return []
+        orm_entity_types = []
+        for entity_type_name in entity_types:
+            entity_type_orm = db_session.query(Entitytype).filter(Entitytype.name == entity_type_name).one_or_none()
+            if entity_type_orm:
+                orm_entity_types.append(entity_type_orm)
+            else:
+                logging.warning("Entity Type  not found: %s.", entity_type_name)
+        return orm_entity_types
+
+    @classmethod
+    def to_orm_from_dict(cls, feed_dict: dict) -> Gtfsrealtimefeed | None:
+        """Convert a dictionary representation of a GTFS RT feed to a SQLAlchemy GtfsRTFeed ORM object."""
+        if not feed_dict:
+            return None
+        feed: FeedOrm = super().to_orm_from_dict(feed_dict)
+        if not feed:
+            return None
+        allowed = {col.name for col in Gtfsrealtimefeed.__mapper__.columns} | {
+            rel.key for rel in Gtfsrealtimefeed.__mapper__.relationships
+        }
+        data = {k: v for k, v in feed.__dict__.items() if k in allowed}
+        result = Gtfsrealtimefeed(**data)
+        result.entitytypes = cls.to_orm_entity_types(feed_dict.get("entity_types"))
+        return result

docs/OperationsAPI.yaml

@@ -95,6 +95,7 @@ paths:
         201:
           description: >
             The feed was successfully created. No content is returned.
+
           content:
             application/json:
               schema:
@@ -189,6 +190,43 @@ paths:
               schema:
                 $ref: "#/components/schemas/OperationGtfsRtFeed"
   /v1/operations/feeds/gtfs_rt:
+    post:
+      description: Create a GTFS-RT feed in the Mobility Database.
+      tags:
+        - "operations"
+      operationId: createGtfsRtFeed
+      security:
+        - ApiKeyAuth: []
+      requestBody:
+        description: Payload to create the specified GTF-RT feed.
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/OperationCreateRequestGtfsRtFeed"
+      responses:
+        201:
+          description: >
+            The feed was successfully created. No content is returned.
+
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/OperationGtfsRtFeed"
+        400:
+          description: >
+            The request was invalid.
+
+        401:
+          description: >
+            The request was not authenticated or has invalid authentication credentials.
+
+        409:
+          description: >
+            A feed with the producer_url already exists.
+
+        500:
+          description: "An internal server error occurred.  \n"  
     put:
       description: Update the specified GTFS-RT feed in the Mobility Database.
       tags:
@@ -1089,6 +1127,81 @@ components:
       required:
         - source_info
         - operational_status
+    OperationCreateRequestGtfsRtFeed:
+      x-operation: true
+      type: object
+      properties:
+        status:
+          $ref: "#/components/schemas/FeedStatus"
+        external_ids:
+          $ref: "#/components/schemas/ExternalIds"
+        provider:
+          description: A commonly used name for the transit provider included in the feed.
+          type: string
+          example: Los Angeles Department of Transportation (LADOT, DASH, Commuter Express)
+        feed_name:
+          description: >
+            An optional description of the data feed, e.g to specify if the data feed is an aggregate of multiple providers, or which network is represented by the feed.
+          type: string
+          example: Bus
+        note:
+          description: A note to clarify complex use cases for consumers.
+          type: string
+        feed_contact_email:
+          description: Use to contact the feed producer.
+          type: string
+          example: [email protected]
+        source_info:
+          allOf:
+            - $ref: "#/components/schemas/SourceInfo"
+            - type: object
+              required:
+                - producer_url
+        redirects:
+          type: array
+          items:
+            $ref: "#/components/schemas/Redirect"
+        operational_status:
+          type: string
+          enum: [wip, published, unpublished]
+          default: wip
+          description: Current operational status of the feed.
+        official:
+          type: boolean
+          description: Whether this is an official feed.
+        entity_types:
+          type: array
+          minItems: 1
+          items:
+            type: string
+            enum:
+              - vp
+              - tu
+              - sa
+            example: vp
+            description: >
+              The type of realtime entry:
+                * vp - vehicle positions
+                * tu - trip updates
+                * sa - service alerts
+        feed_references:
+          description: A list of the GTFS feeds that the real time source is associated with, represented by their MDB source IDs.
+          type: array
+          items:
+            type: string
+            example: "mdb-20"
+        locations:
+          $ref: "#/components/schemas/Locations"
+        related_links:
+          description: >
+            A list of related links for the feed.
+          type: array
+          items:
+            $ref: '#/components/schemas/FeedRelatedLink'          
+      required:
+        - source_info
+        - operational_status
+        - entity_types        
     OperationFeed:
       x-operation: true
       allOf:
@@ -1185,11 +1298,6 @@ components:
             example: vp
             description: >
               The type of realtime entry:
-
-
-
-
-
                 * vp - vehicle positions
                 * tu - trip updates
                 * sa - service alerts
@@ -1265,12 +1373,6 @@ components:
       x-operation: true
       description: >
         Describes status of the Feed. Should be one of
-
-
-
-
-
-
           * `active` Feed should be used in public trip planners.
           * `deprecated` Feed is explicitly deprecated and should not be used in public trip planners.
           * `inactive` Feed hasn't been recently updated and should be used at risk of providing outdated information.
@@ -1288,12 +1390,6 @@ components:
       x-operation: true
       description: >
         Describes data type of a feed. Should be one of
-
-
-
-
-
-
           * `gtfs` GTFS feed.
           * `gtfs_rt` GTFS-RT feed.
           * `gbfs` GBFS feed.

functions-python/operations_api/.openapi-generator/FILES

@@ -26,6 +26,7 @@ src/feeds_gen/models/location.py
 src/feeds_gen/models/metadata.py
 src/feeds_gen/models/operation_create_request_gtfs_feed.py
 src/feeds_gen/models/operation_create_request_gtfs_feed_source_info.py
+src/feeds_gen/models/operation_create_request_gtfs_rt_feed.py
 src/feeds_gen/models/operation_feed.py
 src/feeds_gen/models/operation_gtfs_feed.py
 src/feeds_gen/models/operation_gtfs_rt_feed.py

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

@@ -31,6 +31,9 @@
 from feeds_gen.models.operation_create_request_gtfs_feed import (
     OperationCreateRequestGtfsFeed,
 )
+from feeds_gen.models.operation_create_request_gtfs_rt_feed import (
+    OperationCreateRequestGtfsRtFeed,
+)
 from feeds_gen.models.operation_gtfs_feed import OperationGtfsFeed
 from feeds_gen.models.operation_gtfs_rt_feed import OperationGtfsRtFeed
 from feeds_operations.impl.models.update_request_gtfs_feed_impl import (
@@ -59,6 +62,9 @@
 from .models.operation_create_request_gtfs_feed import (
     OperationCreateRequestGtfsFeedImpl,
 )
+from .models.operation_create_request_gtfs_rt_feed import (
+    OperationCreateRequestGtfsRtFeedImpl,
+)
 from .models.operation_feed_impl import OperationFeedImpl
 from .models.operation_gtfs_feed_impl import OperationGtfsFeedImpl
 from .models.operation_gtfs_rt_feed_impl import OperationGtfsRtFeedImpl
@@ -68,6 +74,37 @@
 class OperationsApiImpl(BaseOperationsApi):
     """Implementation of the operations API."""
 
+    @staticmethod
+    def assign_feed_id(new_feed: Gtfsfeed | Gtfsrealtimefeed):
+        client_provided_id = bool(getattr(new_feed, "id", None))
+        if not client_provided_id:
+            new_feed.id = new_feed.stable_id
+
+    @staticmethod
+    def assign_stable_id(new_feed: Gtfsfeed | Gtfsrealtimefeed, db_session: Session):
+        client_provided_stable_id = bool(getattr(new_feed, "stable_id", None))
+        if not client_provided_stable_id:
+            next_val = db_session.execute(
+                text("SELECT nextval('md_sequence')")
+            ).scalar_one()
+            new_feed.stable_id = f"md-{next_val}"
+
+    @staticmethod
+    def assert_no_existing_feed_url(producer_url: str, db_session: Session):
+        existing_feed = get_feed_by_normalized_url(producer_url, db_session)
+        if existing_feed:
+            message = (
+                f"A published feed with url "
+                f"{producer_url} already exists."
+                f"Existing feed ID: {existing_feed.stable_id}, "
+                f"URL: {existing_feed.producer_url}"
+            )
+            logging.error(message)
+            raise HTTPException(
+                status_code=400,
+                detail=message,
+            )
+
     @with_db_session
     async def get_feeds(
         self,
@@ -314,38 +351,48 @@ async def create_gtfs_feed(
     ) -> OperationGtfsFeed:
         """Create a GTFS feed in the Mobility Database."""
         #         Check if the provider_url already exists in an active feed
-        existing_feed = get_feed_by_normalized_url(
-            operation_create_request_gtfs_feed.source_info.producer_url, db_session
+        OperationsApiImpl.assert_no_existing_feed_url(
+            operation_create_request_gtfs_feed.source_info.producer_url,
+            db_session,
         )
-        if existing_feed:
-            message = (
-                f"A published feed with url "
-                f"{operation_create_request_gtfs_feed.source_info.producer_url} already exists."
-                f"Existing feed ID: {existing_feed.stable_id}, "
-                f"URL: {existing_feed.producer_url}"
-            )
-            logging.error(message)
-            raise HTTPException(
-                status_code=400,
-                detail=message,
-            )
         # Proceed with feed creation
         new_feed = OperationCreateRequestGtfsFeedImpl.to_orm(
             operation_create_request_gtfs_feed
         )
         new_feed.data_type = DataType.GTFS.value
-        client_provided_stable_id = bool(getattr(new_feed, "stable_id", None))
-        if not client_provided_stable_id:
-            next_val = db_session.execute(
-                text("SELECT nextval('md_sequence')")
-            ).scalar_one()
-            new_feed.stable_id = f"md-{next_val}"
-        client_provided_id = bool(getattr(new_feed, "id", None))
-        if not client_provided_id:
-            new_feed.id = new_feed.stable_id
+        OperationsApiImpl.assign_stable_id(new_feed, db_session)
+        OperationsApiImpl.assign_feed_id(new_feed)
         db_session.add(new_feed)
         db_session.commit()
         created_feed = db_session.get(Gtfsfeed, new_feed.id)
         logging.info("Created new GTFS feed with ID: %s", new_feed.stable_id)
         payload = OperationGtfsFeedImpl.from_orm(created_feed).model_dump()
         return JSONResponse(status_code=201, content=jsonable_encoder(payload))
+
+    @with_db_session
+    async def create_gtfs_rt_feed(
+        self,
+        operation_create_request_gtfs_rt_feed: Annotated[
+            OperationCreateRequestGtfsRtFeed,
+            Field(description="Payload to create the specified GTF-RT feed."),
+        ],
+        db_session: Session = None,
+    ) -> OperationGtfsRtFeed:
+        """Create a GTFS-RT feed in the Mobility Database."""
+        OperationsApiImpl.assert_no_existing_feed_url(
+            operation_create_request_gtfs_rt_feed.source_info.producer_url,
+            db_session,
+        )
+        # Proceed with feed creation
+        new_feed = OperationCreateRequestGtfsRtFeedImpl.to_orm(
+            operation_create_request_gtfs_rt_feed
+        )
+        new_feed.data_type = DataType.GTFS_RT.value
+        OperationsApiImpl.assign_stable_id(new_feed, db_session)
+        OperationsApiImpl.assign_feed_id(new_feed)
+        db_session.add(new_feed)
+        db_session.commit()
+        created_feed = db_session.get(Gtfsrealtimefeed, new_feed.id)
+        logging.info("Created new GTFS-RT feed with ID: %s", new_feed.stable_id)
+        payload = OperationGtfsRtFeedImpl.from_orm(created_feed).model_dump()
+        return JSONResponse(status_code=201, content=jsonable_encoder(payload))