feat: JBDA import

api/.openapi-generator/FILES

@@ -18,6 +18,7 @@ src/feeds_gen/models/bounding_box.py
 src/feeds_gen/models/external_id.py
 src/feeds_gen/models/extra_models.py
 src/feeds_gen/models/feed.py
+src/feeds_gen/models/feed_related_link.py
 src/feeds_gen/models/gbfs_endpoint.py
 src/feeds_gen/models/gbfs_feed.py
 src/feeds_gen/models/gbfs_validation_report.py

api/src/feeds/impl/models/feed_impl.py

@@ -1,4 +1,5 @@
 from feeds.impl.models.basic_feed_impl import BaseFeedImpl
+from feeds.impl.models.feed_related_link_impl import FeedRelatedLinkImpl
 from feeds_gen.models.feed import Feed
 from shared.database_gen.sqlacodegen_models import Feed as FeedOrm
 
@@ -24,4 +25,5 @@ def from_orm(cls, feed_orm: FeedOrm | None) -> Feed | None:
         feed.official_updated_at = feed_orm.official_updated_at
         feed.feed_name = feed_orm.feed_name
         feed.note = feed_orm.note
+        feed.related_links = [FeedRelatedLinkImpl.from_orm(related_link) for related_link in feed_orm.feedrelatedlinks]
         return feed

api/src/feeds/impl/models/feed_related_link_impl.py

@@ -0,0 +1,23 @@
+from feeds_gen.models.feed_related_link import FeedRelatedLink
+from shared.database_gen.sqlacodegen_models import Feedrelatedlink
+
+
+class FeedRelatedLinkImpl(FeedRelatedLink):
+    """Implementation of the FeedRelatedLink model."""
+
+    class Config:
+        """Pydantic configuration.
+        Enabling `from_orm` method to create a model instance from a SQLAlchemy row object."""
+
+        from_attributes = True
+
+    @classmethod
+    def from_orm(cls, feed_related_link_orm: Feedrelatedlink) -> FeedRelatedLink | None:
+        if not feed_related_link_orm:
+            return None
+        return cls(
+            code=feed_related_link_orm.code,
+            url=feed_related_link_orm.url,
+            description=feed_related_link_orm.description,
+            created_at=feed_related_link_orm.created_at,
+        )

api/src/shared/common/db_utils.py

@@ -379,6 +379,7 @@ def get_joinedload_options(include_extracted_location_entities: bool = False) ->
     return joinedload_options + [
         joinedload(Feed.locations),
         joinedload(Feed.externalids),
+        joinedload(Feed.feedrelatedlinks),
         joinedload(Feed.redirectingids).joinedload(Redirectingid.target),
         joinedload(Feed.officialstatushistories),
     ]

api/tests/unittest/models/test_basic_feed_impl.py

@@ -95,6 +95,7 @@
     feed_name="feed_name",
     note="note",
     feed_contact_email="feed_contact_email",
+    related_links=[],
     source_info=SourceInfo(
         producer_url="producer_url",
         authentication_type=1,
@@ -149,6 +150,7 @@ def test_from_orm_empty_fields(self):
             external_ids=[],
             redirects=[],
             source_info=SourceInfo(),
+            related_links=[],
         )
         empty_result = FeedImpl.from_orm(empty_feed_orm)
         assert empty_result == expected_empty_feed

api/tests/unittest/models/test_gbfs_feed_impl.py

@@ -49,6 +49,7 @@ def test_from_orm_all_fields(self):
         expected = GbfsFeedImpl(
             id="feed_stable_1",
             system_id="sys1",
+            related_links=[],
             data_type="gbfs",
             created_at=datetime(2024, 1, 1, 10, 0, 0),
             external_ids=[],
@@ -88,6 +89,7 @@ def test_from_orm_empty_fields(self):
             redirects=[],
             locations=[],
             versions=[],
+            related_links=[],
             bounding_box=None,
             bounding_box_generated_at=None,
             source_info=SourceInfo(

api/tests/unittest/models/test_gtfs_feed_impl.py

@@ -134,6 +134,7 @@ def create_test_notice(notice_code: str, total_notices: int, severity: str):
     feed_name="feed_name",
     note="note",
     feed_contact_email="feed_contact_email",
+    related_links=[],
     source_info=SourceInfo(
         producer_url="producer_url",
         authentication_type=1,
@@ -226,6 +227,7 @@ def test_from_orm_empty_fields(self):
             provider=None,
             feed_name="",
             note=None,
+            related_links=[],
             feed_contact_email=None,
             source_info=SourceInfo(
                 producer_url=None,

api/tests/unittest/models/test_gtfs_rt_feed_impl.py

@@ -64,6 +64,7 @@
     id="id",
     data_type="gtfs_rt",
     status="active",
+    related_links=[],
     external_ids=[ExternalIdImpl(external_id="associated_id", source="source")],
     provider="provider",
     feed_name="feed_name",

api/tests/unittest/test_feeds.py

@@ -72,6 +72,7 @@
             "producer_url": "test_producer_url",
         },
         external_ids=[{"external_id": "test_associated_id", "source": "test_source"}],
+        related_links=[],
         redirects=[{"comment": "Some comment", "target_id": "test_target_id"}],
     ).model_dump_json()
 )

docs/DatabaseCatalogAPI.yaml

@@ -454,7 +454,36 @@ components:
             note:
               description: A note to clarify complex use cases for consumers.
               type: string
-
+            related_links:
+              description: >
+                A list of related links for the feed.
+              type: array
+              items:
+                $ref: '#/components/schemas/FeedRelatedLink'
+    FeedRelatedLink:
+      type: object
+      properties:
+        code:
+          description: >
+            A short code to identify the type of link.
+          type: string
+          example: next_1
+        description:
+          description: >
+            A description of the link.
+          type: string
+          example: The next feed URL after the current one
+        url:
+          description: >
+            The URL of the related link.
+          type: string
+          format: url
+        created_at:
+          description: >
+            The date and time the related link was created, in ISO 8601 date-time format.
+          type: string
+          example: 2023-07-10T22:06:00Z
+          format: date-time
     GtfsFeed:
       allOf:
         - $ref: "#/components/schemas/Feed"

functions-python/helpers/locations.py

@@ -113,6 +113,7 @@ def create_or_get_location(
         municipality=city_name,
     )
     session.add(location)
+    session.flush()
     logging.debug(f"Created new location: {location_id}")
 
     return location

functions-python/tasks_executor/requirements.txt

@@ -27,3 +27,4 @@ google-cloud-storage
 
 # Configuration
 python-dotenv==1.0.0
+pycountry

functions-python/tasks_executor/src/main.py

@@ -38,6 +38,7 @@
 from tasks.geojson.update_geojson_files_precision import (
     update_geojson_files_precision_handler,
 )
+from tasks.data_import.import_jbda_feeds import import_jbda_handler
 
 init_logger()
 LIST_COMMAND: Final[str] = "list"
@@ -77,6 +78,10 @@
         "description": "Rebuilds missing visualization files for GTFS datasets.",
         "handler": rebuild_missing_visualization_files_handler,
     },
+    "jbda_import": {
+        "description": "Imports JBDA data into the system.",
+        "handler": import_jbda_handler,
+    },
 }
 
 

functions-python/tasks_executor/src/tasks/data_import/data_import_utils.py

@@ -0,0 +1,43 @@
+import os
+from shared.database_gen.sqlacodegen_models import Feed
+import logging
+import json
+from google.cloud import pubsub_v1
+
+PROJECT_ID = os.getenv("PROJECT_ID")
+DATASET_BATCH_TOPIC = os.getenv("DATASET_PROCESSING_TOPIC_NAME")
+
+
+def trigger_dataset_download(
+    feed: Feed, execution_id: str, publisher: pubsub_v1.PublisherClient
+) -> None:
+    """Publishes the feed to the configured Pub/Sub topic."""
+    topic_path = publisher.topic_path(PROJECT_ID, DATASET_BATCH_TOPIC)
+    logging.debug(f"Publishing to Pub/Sub topic: {topic_path}")
+
+    message_data = {
+        "execution_id": execution_id,
+        "producer_url": feed.producer_url,
+        "feed_stable_id": feed.stable_id,
+        "feed_id": feed.id,
+        "dataset_id": None,
+        "dataset_hash": None,
+        "authentication_type": feed.authentication_type,
+        "authentication_info_url": feed.authentication_info_url,
+        "api_key_parameter_name": feed.api_key_parameter_name,
+    }
+
+    try:
+        # Convert to JSON string
+        json_message = json.dumps(message_data)
+        future = publisher.publish(topic_path, data=json_message.encode("utf-8"))
+        future.add_done_callback(
+            lambda _: logging.info(
+                "Published feed %s to dataset batch topic", feed.stable_id
+            )
+        )
+        future.result()
+        logging.info("Message published for feed %s", feed.stable_id)
+    except Exception as e:
+        logging.error("Error publishing to dataset batch topic: %s", str(e))
+        raise