diff --git a/.github/workflows/api-deployer.yml b/.github/workflows/api-deployer.yml index cfc8310be..71b6e14d2 100644 --- a/.github/workflows/api-deployer.yml +++ b/.github/workflows/api-deployer.yml @@ -207,7 +207,7 @@ jobs: echo "POSTGRES_DB=${{ env.local_postgres_db }}" >> config/.env.local echo "POSTGRES_PORT=${{ env.local_postgres_port }}" >> config/.env.local echo "POSTGRES_HOST=localhost" >> config/.env.local - echo "ENV=dev" >> config/.env.local + echo "ENVIRONMENT=dev" >> config/.env.local # db models were generated and uploaded in api-build-test job above. - uses: actions/download-artifact@v4 diff --git a/.github/workflows/db-update-content.yml b/.github/workflows/db-update-content.yml index 07e25855b..69e1be340 100644 --- a/.github/workflows/db-update-content.yml +++ b/.github/workflows/db-update-content.yml @@ -95,7 +95,7 @@ jobs: echo "FEEDS_DATABASE_URL=postgresql://${{ secrets.DB_USER_NAME }}:${{ secrets.DB_USER_PASSWORD }}@localhost:5432/${{ inputs.DB_NAME }}" >> config/.env.local echo "POSTGRES_PORT=5432" >> config/.env.local echo "POSTGRES_HOST=localhost" >> config/.env.local - echo "ENV=${{ inputs.ENVIRONMENT }}" >> config/.env.local + echo "ENVIRONMENT=${{ inputs.ENVIRONMENT }}" >> config/.env.local cat config/.env.local - name: Load secrets from 1Password diff --git a/api/src/scripts/load_dataset_on_create.py b/api/src/scripts/load_dataset_on_create.py index 5d97e641e..d4be55def 100644 --- a/api/src/scripts/load_dataset_on_create.py +++ b/api/src/scripts/load_dataset_on_create.py @@ -29,7 +29,7 @@ def get_pubsub_client(): def get_topic_path(): - env = os.getenv("ENV", "dev") + env = os.getenv("ENVIRONMENT", "dev") pubsub_topic_name = f"datasets-batch-topic-{env}" project_id = f"mobility-feeds-{env}" # Cannot use GOOGLE_CLOUD_PROJECT because it points to QA for DEV return get_pubsub_client().topic_path(project_id, pubsub_topic_name) diff --git a/api/src/scripts/populate_db_gbfs.py b/api/src/scripts/populate_db_gbfs.py index 785f4e418..9edfa22c1 100644 --- a/api/src/scripts/populate_db_gbfs.py +++ b/api/src/scripts/populate_db_gbfs.py @@ -152,14 +152,14 @@ def trigger_downstream_tasks(self): f"Triggering downstream tasks for {len(self.added_feeds)} newly added feed(s): " f"{', '.join(f['stable_id'] for f in self.added_feeds)}" ) - if os.getenv("ENV", "local") == "local": + if os.getenv("ENVIRONMENT", "local") == "local": self.logger.info("Skipping downstream tasks in local environment.") return if not self.added_feeds: self.logger.info("No feeds to trigger downstream tasks for.") return - env = os.getenv("ENV", "dev") + env = os.getenv("ENVIRONMENT", "dev") project_id = f"mobility-feeds-{env}" publisher = pubsub_v1.PublisherClient() topic_path = publisher.topic_path(project_id, GBFS_PUBSUB_TOPIC_NAME) diff --git a/api/src/scripts/populate_db_gtfs.py b/api/src/scripts/populate_db_gtfs.py index 924298660..528aa7dbc 100644 --- a/api/src/scripts/populate_db_gtfs.py +++ b/api/src/scripts/populate_db_gtfs.py @@ -283,9 +283,9 @@ def trigger_downstream_tasks(self): f"{','.join([feed.stable_id for feed in self.added_gtfs_feeds] if self.added_gtfs_feeds else [])}" ) - env = os.getenv("ENV") - self.logger.info(f"ENV = {env}") - if os.getenv("ENV", "local") != "local": + env = os.getenv("ENVIRONMENT") + self.logger.info(f"ENVIRONMENT = {env}") + if os.getenv("ENVIRONMENT", "local") != "local": # Publishes the new feeds to the Pub/Sub topic to download the datasets publish_all(self.added_gtfs_feeds) diff --git a/config/.env.local b/config/.env.local index 3501826d8..ebe020fd3 100644 --- a/config/.env.local +++ b/config/.env.local @@ -6,7 +6,7 @@ POSTGRES_PORT=5432 POSTGRES_TEST_PORT=54320 PGUSER=postgres POSTGRES_HOST=localhost -ENV=local +ENVIRONMENT=local SCHEMA_SPY_DOC=schemaspy-dev FEEDS_DATABASE_URL=postgresql://postgres:postgres@localhost:5432/MobilityDatabase FEEDS_DATABASE_URL_TEST=postgresql://postgres:postgres@localhost:54320/MobilityDatabaseTest diff --git a/functions-python/backfill_dataset_service_date_range/.env.rename_me b/functions-python/backfill_dataset_service_date_range/.env.rename_me index 068e688c4..e9963004d 100644 --- a/functions-python/backfill_dataset_service_date_range/.env.rename_me +++ b/functions-python/backfill_dataset_service_date_range/.env.rename_me @@ -1,3 +1,3 @@ # Environment variables for the validation report information extraction to run locally export FEEDS_DATABASE_URL=${{FEEDS_DATABASE_URL}} -export ENV=${{ENV}} +export ENVIRONMENT=${{ENVIRONMENT}} diff --git a/functions-python/backfill_dataset_service_date_range/src/main.py b/functions-python/backfill_dataset_service_date_range/src/main.py index d40a5c701..885b6285a 100644 --- a/functions-python/backfill_dataset_service_date_range/src/main.py +++ b/functions-python/backfill_dataset_service_date_range/src/main.py @@ -25,7 +25,7 @@ from google.cloud import storage import requests -env = os.getenv("ENV", "dev").lower() +env = os.getenv("ENVIRONMENT", "dev").lower() bucket_name = f"mobilitydata-datasets-{env}" init_logger() diff --git a/functions-python/helpers/validation_report/validation_report_update.py b/functions-python/helpers/validation_report/validation_report_update.py index ff5beb5c1..f5cf87872 100644 --- a/functions-python/helpers/validation_report/validation_report_update.py +++ b/functions-python/helpers/validation_report/validation_report_update.py @@ -7,7 +7,7 @@ from google.cloud.workflows import executions_v1 from google.cloud.workflows.executions_v1 import Execution -env = os.getenv("ENV", "dev").lower() +env = os.getenv("ENVIRONMENT", "dev").lower() bucket_name = f"mobilitydata-datasets-{env}" diff --git a/functions-python/tasks_executor/src/tasks/data_import/transportdatagouv/import_tdg_feeds.py b/functions-python/tasks_executor/src/tasks/data_import/transportdatagouv/import_tdg_feeds.py index acdb44c0b..e4f517705 100644 --- a/functions-python/tasks_executor/src/tasks/data_import/transportdatagouv/import_tdg_feeds.py +++ b/functions-python/tasks_executor/src/tasks/data_import/transportdatagouv/import_tdg_feeds.py @@ -804,7 +804,7 @@ def commit_changes( logger.info("Commit after processing items (count=%d)", total_processed) db_session.commit() execution_id = str(uuid.uuid4()) - if os.getenv("ENV", "").lower() == "local": + if os.getenv("ENVIRONMENT", "").lower() == "local": return for feed in feeds_to_publish: trigger_dataset_download(feed, execution_id) diff --git a/functions-python/tasks_executor/src/tasks/validation_reports/README.md b/functions-python/tasks_executor/src/tasks/validation_reports/README.md index 83dcfbee5..00b1b1a7e 100644 --- a/functions-python/tasks_executor/src/tasks/validation_reports/README.md +++ b/functions-python/tasks_executor/src/tasks/validation_reports/README.md @@ -77,7 +77,8 @@ that were already triggered (tracked in `task_execution_log`). "filter_statuses": ["active"], "filter_op_statuses": ["published"], "force_update": false, - "limit": 10 + "limit": 10, + "reports_bucket_name": "stg-gtfs-validator-results" } ``` @@ -91,6 +92,7 @@ that were already triggered (tracked in `task_execution_log`). | `filter_op_statuses` | list[str] | `["published"]` | Filter feeds by operational status. Accepted values: `"published"`, `"unpublished"`, `"wip"` | | `force_update` | bool | `false` | Re-trigger even when a current report already exists | | `limit` | int | `null` | Cap the number of workflows triggered per call — useful for end-to-end testing | +| `reports_bucket_name` | string | env-derived | Override the GCS bucket where validator results are stored. Use when running in prod but pointing to the staging validator (e.g. `"stg-gtfs-validator-results"`) | --- diff --git a/functions-python/tasks_executor/src/tasks/validation_reports/rebuild_missing_validation_reports.py b/functions-python/tasks_executor/src/tasks/validation_reports/rebuild_missing_validation_reports.py index 8ddf020e8..88c3a3415 100644 --- a/functions-python/tasks_executor/src/tasks/validation_reports/rebuild_missing_validation_reports.py +++ b/functions-python/tasks_executor/src/tasks/validation_reports/rebuild_missing_validation_reports.py @@ -68,6 +68,9 @@ def rebuild_missing_validation_reports_handler(payload) -> dict: Default: False "force_update": bool, # [optional] Re-trigger even if a report already exists. Default: False "limit": int, # [optional] Max datasets to trigger per call (for testing). Default: unlimited + "reports_bucket_name": str, # [optional] Override the GCS bucket for validator results. + # Defaults to the env-derived bucket (prod or staging). + # Use when running in prod but pointing to the staging validator. } """ ( @@ -80,6 +83,7 @@ def rebuild_missing_validation_reports_handler(payload) -> dict: bypass_db_update, force_update, limit, + reports_bucket_name, ) = get_parameters(payload) return rebuild_missing_validation_reports( @@ -92,6 +96,7 @@ def rebuild_missing_validation_reports_handler(payload) -> dict: prod_env=prod_env, force_update=force_update, limit=limit, + reports_bucket_name=reports_bucket_name, ) @@ -106,6 +111,7 @@ def rebuild_missing_validation_reports( prod_env: bool = False, force_update: bool = False, limit: Optional[int] = None, + reports_bucket_name: Optional[str] = None, db_session: Session | None = None, ) -> dict: """ @@ -124,6 +130,8 @@ def rebuild_missing_validation_reports( prod_env: True if targeting the production environment. Default: False force_update: Re-trigger even if a report already exists. Default: False limit: Max datasets to trigger per call (for end-to-end testing). Default: unlimited + reports_bucket_name: Override the GCS bucket for validator results. + Defaults to the env-derived bucket (prod or staging). db_session: DB session (injected by @with_db_session) """ validator_version = _get_validator_version(validator_endpoint) @@ -192,7 +200,8 @@ def rebuild_missing_validation_reports( datasets_to_trigger, validator_endpoint=validator_endpoint, bypass_db_update=bypass_db_update, - reports_bucket_name=get_gtfs_validator_results_bucket(prod_env), + reports_bucket_name=reports_bucket_name + or get_gtfs_validator_results_bucket(prod_env), tracker=tracker, ) total_triggered = len(triggered_ids) @@ -222,6 +231,8 @@ def rebuild_missing_validation_reports( "prod_env": prod_env, "force_update": force_update, "limit": limit, + "reports_bucket_name": reports_bucket_name + or get_gtfs_validator_results_bucket(prod_env), }, } logging.info(result) @@ -322,8 +333,9 @@ def get_parameters(payload): Args: payload (dict): Task payload dict. Returns: - Tuple of (dry_run, filter_after_in_days, filter_statuses, prod_env, - validator_endpoint, force_update, limit) + Tuple of (dry_run, filter_after_in_days, filter_statuses, filter_op_statuses, + prod_env, validator_endpoint, bypass_db_update, force_update, limit, + reports_bucket_name) """ prod_env = os.getenv("ENVIRONMENT", "").lower() == "prod" default_endpoint = get_gtfs_validator_url(prod_env) @@ -363,6 +375,8 @@ def get_parameters(payload): if limit is not None: limit = int(limit) + reports_bucket_name = payload.get("reports_bucket_name", None) + return ( dry_run, filter_after_in_days, @@ -373,4 +387,5 @@ def get_parameters(payload): bypass_db_update, force_update, limit, + reports_bucket_name, ) diff --git a/functions-python/tasks_executor/tests/tasks/data_import/test_tdg_import.py b/functions-python/tasks_executor/tests/tasks/data_import/test_tdg_import.py index e79dbe411..d4a37fa1e 100644 --- a/functions-python/tasks_executor/tests/tasks/data_import/test_tdg_import.py +++ b/functions-python/tasks_executor/tests/tasks/data_import/test_tdg_import.py @@ -207,7 +207,7 @@ def test_import_creates_gtfs_and_rt(self, db_session: Session): mock_trigger, ), patch.dict( os.environ, - {"COMMIT_BATCH_SIZE": "1", "ENV": "test"}, + {"COMMIT_BATCH_SIZE": "1", "ENVIRONMENT": "test"}, clear=False, ): result = import_tdg_handler({"dry_run": False}) diff --git a/functions-python/tasks_executor/tests/tasks/validation_reports/test_rebuild_missing_validation_reports.py b/functions-python/tasks_executor/tests/tasks/validation_reports/test_rebuild_missing_validation_reports.py index c087474df..3440a923e 100644 --- a/functions-python/tasks_executor/tests/tasks/validation_reports/test_rebuild_missing_validation_reports.py +++ b/functions-python/tasks_executor/tests/tasks/validation_reports/test_rebuild_missing_validation_reports.py @@ -39,6 +39,7 @@ def test_defaults(self): bypass_db_update, force_update, limit, + reports_bucket_name, ) = get_parameters({}) self.assertTrue(dry_run) self.assertIsNone(filter_after_in_days) @@ -49,6 +50,7 @@ def test_defaults(self): self.assertFalse(bypass_db_update) self.assertFalse(force_update) self.assertIsNone(limit) + self.assertIsNone(reports_bucket_name) def test_all_params(self): payload = { @@ -60,6 +62,7 @@ def test_all_params(self): "bypass_db_update": True, "force_update": True, "limit": 10, + "reports_bucket_name": "my-custom-bucket", } ( dry_run, @@ -71,6 +74,7 @@ def test_all_params(self): bypass_db_update, force_update, limit, + reports_bucket_name, ) = get_parameters(payload) self.assertFalse(dry_run) self.assertEqual(filter_after_in_days, 30) @@ -80,6 +84,7 @@ def test_all_params(self): self.assertTrue(bypass_db_update) self.assertTrue(force_update) self.assertEqual(limit, 10) + self.assertEqual(reports_bucket_name, "my-custom-bucket") def test_string_coercion(self): payload = { @@ -88,9 +93,18 @@ def test_string_coercion(self): "force_update": "true", "limit": "5", } - dry_run, _, _, _, _, _, bypass_db_update, force_update, limit = get_parameters( - payload - ) + ( + dry_run, + _, + _, + _, + _, + _, + bypass_db_update, + force_update, + limit, + _, + ) = get_parameters(payload) self.assertFalse(dry_run) self.assertTrue(bypass_db_update) self.assertTrue(force_update) @@ -186,6 +200,45 @@ def test_limit_slices_datasets( self.assertEqual(result["total_candidates"], 20) self.assertEqual(result["total_in_call"], 5) + @patch(f"{_MODULE}._get_validator_version", return_value="7.0.0") + @patch(f"{_MODULE}._filter_out_datasets_without_blob", return_value=[("f", "ds-1")]) + @patch(f"{_MODULE}.execute_workflows", return_value=["ds-1"]) + @patch(f"{_MODULE}.TaskExecutionTracker") + def test_reports_bucket_name_override_is_used( + self, tracker_cls, exec_mock, filter_blob_mock, version_mock + ): + """When reports_bucket_name is provided, it overrides the env-derived bucket.""" + session = self._make_session_mock(datasets=[("f", "ds-1")]) + rebuild_missing_validation_reports( + validator_endpoint="https://staging.example.com/api", + dry_run=False, + reports_bucket_name="stg-gtfs-validator-results", + prod_env=True, + db_session=session, + ) + _, call_kwargs = exec_mock.call_args + self.assertEqual( + call_kwargs["reports_bucket_name"], "stg-gtfs-validator-results" + ) + + @patch(f"{_MODULE}._get_validator_version", return_value="7.0.0") + @patch(f"{_MODULE}._filter_out_datasets_without_blob", return_value=[("f", "ds-1")]) + @patch(f"{_MODULE}.execute_workflows", return_value=["ds-1"]) + @patch(f"{_MODULE}.TaskExecutionTracker") + def test_reports_bucket_name_defaults_to_env_derived( + self, tracker_cls, exec_mock, filter_blob_mock, version_mock + ): + """When reports_bucket_name is not set, the env-derived prod bucket is used.""" + session = self._make_session_mock(datasets=[("f", "ds-1")]) + rebuild_missing_validation_reports( + validator_endpoint="https://staging.example.com/api", + dry_run=False, + prod_env=True, + db_session=session, + ) + _, call_kwargs = exec_mock.call_args + self.assertEqual(call_kwargs["reports_bucket_name"], "gtfs-validator-results") + @patch(f"{_MODULE}._get_validator_version", return_value="7.0.0") @patch(f"{_MODULE}._filter_out_datasets_without_blob", return_value=[("f", "ds-1")]) @patch(f"{_MODULE}.execute_workflows", return_value=["ds-1"]) @@ -229,6 +282,7 @@ def test_handler_passes_all_params(self, rebuild_mock): "force_update": True, "limit": 10, "filter_op_statuses": ["published", "wip"], + "reports_bucket_name": "my-custom-bucket", } rebuild_missing_validation_reports_handler(payload) rebuild_mock.assert_called_once_with( @@ -241,6 +295,7 @@ def test_handler_passes_all_params(self, rebuild_mock): prod_env=False, force_update=True, limit=10, + reports_bucket_name="my-custom-bucket", ) @patch(f"{_MODULE}._get_validator_version", return_value="7.0.0") diff --git a/infra/functions-python/main.tf b/infra/functions-python/main.tf index 129226320..994ba9fde 100644 --- a/infra/functions-python/main.tf +++ b/infra/functions-python/main.tf @@ -340,8 +340,6 @@ resource "google_cloudfunctions2_function" "process_validation_report" { vpc_connector_egress_settings = "PRIVATE_RANGES_ONLY" environment_variables = { - # FIXME, duplicate variable - ENV = var.environment ENVIRONMENT = var.environment PROJECT_ID = var.project_id GCP_REGION = var.gcp_region @@ -395,7 +393,7 @@ resource "google_cloudfunctions2_function" "compute_validation_report_counters" vpc_connector_egress_settings = "PRIVATE_RANGES_ONLY" environment_variables = { - ENV = var.environment + ENVIRONMENT = var.environment PYTHONNODEBUGRANGES = 0 } @@ -596,7 +594,7 @@ resource "google_cloudfunctions2_function" "gbfs_validator_pubsub" { vpc_connector = data.google_vpc_access_connector.vpc_connector.id vpc_connector_egress_settings = "PRIVATE_RANGES_ONLY" environment_variables = { - ENV = var.environment + ENVIRONMENT = var.environment BUCKET_NAME = google_storage_bucket.gbfs_snapshots_bucket.name PROJECT_ID = var.project_id GCP_REGION = var.gcp_region @@ -688,7 +686,7 @@ resource "google_cloudfunctions2_function" "backfill_dataset_service_date_range" environment_variables = { # prevents multiline logs from being truncated on GCP console - ENV = var.environment + ENVIRONMENT = var.environment PYTHONNODEBUGRANGES = 0 } dynamic "secret_environment_variables" { @@ -1151,7 +1149,7 @@ resource "google_cloudfunctions2_function" "pmtiles_builder" { service_config { environment_variables = { PROJECT_ID = var.project_id - ENV = var.environment + ENVIRONMENT = var.environment PUBSUB_TOPIC_NAME = "rebuild-bounding-boxes-topic" MATERIALIZED_VIEW_QUEUE = google_cloud_tasks_queue.refresh_materialized_view_task_queue.name DATASETS_BUCKET_NAME = "${var.datasets_bucket_name}-${var.environment}"