feat: Transport Data Gouv feeds fetching (#1478)

Author: cka-y

.github/workflows/api-deployer.yml

@@ -297,12 +297,13 @@ jobs:
         env:
           OP_SERVICE_ACCOUNT_TOKEN: ${{ secrets.OP_SERVICE_ACCOUNT_TOKEN }}
           TRANSITLAND_API_KEY: "op://rbiv7rvkkrsdlpcrz3bmv7nmcu/TansitLand API Key/credential"
+          TDG_API_TOKEN: "op://rbiv7rvkkrsdlpcrz3bmv7nmcu/Transport.data.gouv.fr API Token/credential"
           OPERATIONS_OAUTH2_CLIENT_ID: ${{ inputs.OPERATIONS_OAUTH2_CLIENT_ID_1PASSWORD }}
 
       - name: Populate Variables
         run: |
           scripts/replace-variables.sh -in_file infra/backend.conf.rename_me -out_file infra/backend.conf -variables BUCKET_NAME,OBJECT_PREFIX
-          scripts/replace-variables.sh -in_file infra/vars.tfvars.rename_me -out_file infra/vars.tfvars -variables PROJECT_ID,REGION,ENVIRONMENT,DEPLOYER_SERVICE_ACCOUNT,FEED_API_IMAGE_VERSION,OAUTH2_CLIENT_ID,OAUTH2_CLIENT_SECRET,GLOBAL_RATE_LIMIT_REQ_PER_MINUTE,ARTIFACT_REPO_NAME,VALIDATOR_ENDPOINT,TRANSITLAND_API_KEY,OPERATIONS_OAUTH2_CLIENT_ID
+          scripts/replace-variables.sh -in_file infra/vars.tfvars.rename_me -out_file infra/vars.tfvars -variables PROJECT_ID,REGION,ENVIRONMENT,DEPLOYER_SERVICE_ACCOUNT,FEED_API_IMAGE_VERSION,OAUTH2_CLIENT_ID,OAUTH2_CLIENT_SECRET,GLOBAL_RATE_LIMIT_REQ_PER_MINUTE,ARTIFACT_REPO_NAME,VALIDATOR_ENDPOINT,TRANSITLAND_API_KEY,OPERATIONS_OAUTH2_CLIENT_ID,TDG_API_TOKEN
 
       - uses: hashicorp/setup-terraform@v3
         with:

functions-python/tasks_executor/function_config.json

@@ -10,6 +10,9 @@
   "environment_variables": [
     {
       "key": "DATASETS_BUCKET_NAME"
+    },
+    {
+      "key": "TDG_API_TOKEN"
     }
   ],
   "secret_environment_variables": [

functions-python/tasks_executor/src/main.py

@@ -22,6 +22,7 @@
 
 from shared.helpers.logger import init_logger
 from tasks.data_import.transitfeeds.sync_transitfeeds import sync_transitfeeds_handler
+from tasks.data_import.transportdatagouv.import_tdg_feeds import import_tdg_handler
 from tasks.dataset_files.rebuild_missing_dataset_files import (
     rebuild_missing_dataset_files_handler,
 )
@@ -109,6 +110,10 @@
         "description": "Syncs data from TransitFeeds to the database.",
         "handler": sync_transitfeeds_handler,
     },
+    "tdg_import": {
+        "description": "Imports TDG data into the system.",
+        "handler": import_tdg_handler,
+    },
 }
 
 

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

@@ -1,7 +1,7 @@
 import logging
 import uuid
 from datetime import datetime
-from typing import Tuple, Type, TypeVar
+from typing import Tuple, Type, TypeVar, Optional
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -10,13 +10,14 @@
     Feed,
     Officialstatushistory,
     Entitytype,
+    License,
 )
 
 logger = logging.getLogger(__name__)
 T = TypeVar("T", bound="Feed")
 
 
-def _get_or_create_entity_type(session: Session, entity_type_name: str) -> Entitytype:
+def get_or_create_entity_type(session: Session, entity_type_name: str) -> Entitytype:
     """Get or create an Entitytype by name."""
     logger.debug("Looking up Entitytype name=%s", entity_type_name)
     et = session.scalar(select(Entitytype).where(Entitytype.name == entity_type_name))
@@ -45,7 +46,21 @@ def get_feed(
     return feed
 
 
-def _get_or_create_feed(
+def get_license(session: Session, license_id: str) -> Optional[License]:
+    """Get a License by ID."""
+    logger.debug("Lookup License id=%s", license_id)
+    if not license_id:
+        logger.debug("No License ID provided")
+        return None
+    license = session.get(License, license_id)
+    if license:
+        logger.debug("Found existing License id=%s", license_id)
+        return license
+    logger.debug("No License found with id=%s", license_id)
+    return None
+
+
+def get_or_create_feed(
     session: Session,
     model: Type[T],
     stable_id: str,

functions-python/tasks_executor/src/tasks/data_import/jbda/import_jbda_feeds.py

@@ -39,8 +39,8 @@
 )
 from shared.helpers.pub_sub import trigger_dataset_download
 from tasks.data_import.data_import_utils import (
-    _get_or_create_entity_type,
-    _get_or_create_feed,
+    get_or_create_entity_type,
+    get_or_create_feed,
 )
 
 T = TypeVar("T", bound="Feed")
@@ -307,9 +307,9 @@ def _upsert_rt_feeds(
             )
             continue
 
-        et = _get_or_create_entity_type(db_session, entity_type_name)
+        et = get_or_create_entity_type(db_session, entity_type_name)
         rt_stable_id = f"{stable_id}-{entity_type_name}"
-        rt_feed, is_new_rt = _get_or_create_feed(
+        rt_feed, is_new_rt = get_or_create_feed(
             db_session, Gtfsrealtimefeed, rt_stable_id, "gtfs_rt"
         )
 
@@ -388,9 +388,7 @@ def _process_feed(
 
     # Upsert/lookup schedule feed
     stable_id = f"jbda-{org_id}-{feed_id}"
-    gtfs_feed, is_new_gtfs = _get_or_create_feed(
-        db_session, Gtfsfeed, stable_id, "gtfs"
-    )
+    gtfs_feed, is_new_gtfs = get_or_create_feed(db_session, Gtfsfeed, stable_id, "gtfs")
 
     # Diff detection
     api_sched_fp = _build_api_schedule_fingerprint(item, dbody, producer_url)

functions-python/tasks_executor/src/tasks/data_import/transitfeeds/sync_transitfeeds.py

@@ -34,8 +34,8 @@
     Gtfsdataset,
 )
 from tasks.data_import.data_import_utils import (
-    _get_or_create_feed,
-    _get_or_create_entity_type,
+    get_or_create_feed,
+    get_or_create_entity_type,
     get_feed,
 )
 
@@ -80,7 +80,7 @@ def _process_feeds(
                 feed_stable_id,
             )
 
-            feed, is_new = _get_or_create_feed(
+            feed, is_new = get_or_create_feed(
                 db_session, model_cls, feed_stable_id, feed_kind, is_official=False
             )
             # All TransitFeeds imports are marked deprecated
@@ -274,7 +274,7 @@ def _rt_on_is_new(session: Session, feed, row: pd.Series) -> None:
         )
         if entity_types:
             feed.entitytypes = [
-                _get_or_create_entity_type(session, et) for et in entity_types
+                get_or_create_entity_type(session, et) for et in entity_types
             ]
             logger.info(
                 "[GTFS_RT] Set %d entity types for %s",