Added global and per feed configuration

Author: qcdyx

api/src/utils/config_reader.py

@@ -0,0 +1,62 @@
+from typing import Optional, Any
+from sqlalchemy.orm import Session
+from shared.database.database import with_db_session
+from shared.database_gen.sqlacodegen_models import ConfigKey, ConfigValueFeed
+
+
+@with_db_session
+def get_config_value(
+    namespace: str,
+    key: str,
+    feed_id: Optional[str] = None,
+    db_session: Session = None,
+) -> Optional[Any]:
+    """
+    Retrieves a configuration value from the database using the provided session.
+
+    It first looks for a feed-specific override. If a feed_id is provided and a
+    value is found, it returns that value.
+
+    If no feed-specific value is found or no feed_id is provided, it looks for
+    the global default value in the `config_key` table.
+
+    :param namespace: The namespace of the configuration key.
+    :param key: The configuration key.
+    :param feed_id: The optional feed_id for a specific override.
+    :param db_session: The SQLAlchemy session, injected by the `with_db_session` decorator.
+    :return: The configuration value, or None if not found.
+    """
+    # 1. Try to get feed-specific value if feed_id is provided
+    if feed_id:
+        feed_config = (
+            db_session.query(ConfigValueFeed.value)
+            .filter(
+                ConfigValueFeed.feed_id == feed_id,
+                ConfigValueFeed.namespace == namespace,
+                ConfigValueFeed.key == key,
+            )
+            .first()
+        )
+        if feed_config:
+            return feed_config.value
+
+    # 2. If not found or no feed_id, get the default value
+    default_config = (
+        db_session.query(ConfigKey.default_value).filter(ConfigKey.namespace == namespace, ConfigKey.key == key).first()
+    )
+
+    return default_config.default_value if default_config else None
+
+
+# Example of how you would use it elsewhere in your application:
+#
+# from shared.config.config_reader import get_config_value
+#
+# def my_service_function():
+#     # The `with_db_session` decorator handles creating and closing the session.
+#     http_headers = get_config_value(
+#         namespace='feed_download',
+#         key='http_headers',
+#         feed_id='mdb-2126'
+#     )
+#     print(f"HTTP Headers for mdb-2126: {http_headers}")

functions-python/batch_process_dataset/src/main.py

@@ -122,13 +122,14 @@ def create_dataset_stable_id(feed_stable_id, timestamp):
         """
         return f"{feed_stable_id}-{timestamp}"
 
-    def download_content(self, temporary_file_path):
+    def download_content(self, temporary_file_path, stable_id):
         """
         Downloads the content of a URL and return the hash of the file
         """
         file_hash = download_and_get_hash(
             self.producer_url,
             temporary_file_path,
+            stable_id,
             authentication_type=self.authentication_type,
             api_key_parameter_name=self.api_key_parameter_name,
             credentials=self.feed_credentials,
@@ -193,7 +194,7 @@ def upload_files_to_storage(
                 )
         return blob, extracted_files
 
-    def upload_dataset(self, public=True) -> DatasetFile or None:
+    def upload_dataset(self, stable_id, public=True) -> DatasetFile or None:
         """
         Uploads a dataset to a GCP bucket as <feed_stable_id>/latest.zip and
         <feed_stable_id>/<feed_stable_id>-<upload_datetime>.zip
@@ -204,7 +205,7 @@ def upload_dataset(self, public=True) -> DatasetFile or None:
         try:
             self.logger.info("Accessing URL %s", self.producer_url)
             temp_file_path = self.generate_temp_filename()
-            file_sha256_hash, is_zip = self.download_content(temp_file_path)
+            file_sha256_hash, is_zip = self.download_content(temp_file_path, stable_id)
             if not is_zip:
                 self.logger.error(
                     f"[{self.feed_stable_id}] The downloaded file from {self.producer_url} is not a valid ZIP file."
@@ -416,12 +417,12 @@ def _get_unzipped_size(dataset_file):
         )
 
     @with_db_session
-    def process_from_producer_url(self, db_session) -> Optional[DatasetFile]:
+    def process_from_producer_url(self, db_session, stable_id) -> Optional[DatasetFile]:
         """
         Process the dataset and store new version in GCP bucket if any changes are detected
         :return: the DatasetFile object created
         """
-        dataset_file = self.upload_dataset()
+        dataset_file = self.upload_dataset(stable_id)
 
         if dataset_file is None:
             self.logger.info(f"[{self.feed_stable_id}] No database update required.")
@@ -542,7 +543,7 @@ def process_dataset(cloud_event: CloudEvent):
         if json_payload.get("use_bucket_latest", False):
             dataset_file = processor.process_from_bucket()
         else:
-            dataset_file = processor.process_from_producer_url()
+            dataset_file = processor.process_from_producer_url(stable_id)
     except Exception as e:
         # This makes sure the logger is initialized
         logger = get_logger("process_dataset", stable_id if stable_id else "UNKNOWN")

functions-python/helpers/utils.py

@@ -28,6 +28,8 @@
 from urllib3.util.ssl_ import create_urllib3_context
 from pathlib import Path
 
+from utils.config_reader import get_config_value
+
 
 def create_bucket(bucket_name):
     """
@@ -126,6 +128,7 @@ def download_and_get_hash(
     file_path,
     hash_algorithm="sha256",
     chunk_size=8192,
+    stable_id=None,
     authentication_type=0,
     api_key_parameter_name=None,
     credentials=None,
@@ -145,12 +148,17 @@ def download_and_get_hash(
         ctx.load_default_certs()
         ctx.options |= 0x4  # ssl.OP_LEGACY_SERVER_CONNECT
 
+        headers = get_config_value(
+            namespace="feed_download", key="http_headers", feed_stable_id=stable_id
+        )
+        if headers is None:
+            headers = {
+                "User-Agent": "Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) "
+                "AppleWebKit/537.36 (KHTML, like Gecko) "
+                "Chrome/126.0.0.0 Mobile Safari/537.36"
+            }
+
         # authentication_type == 1 -> the credentials are passed in the url
-        headers = {
-            "User-Agent": "Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) "
-            "AppleWebKit/537.36 (KHTML, like Gecko) "
-            "Chrome/126.0.0.0 Mobile Safari/537.36"
-        }
         # Careful, some URLs may already contain a query string
         # (e.g. http://api.511.org/transit/datafeeds?operator_id=CE)
         if authentication_type == 1 and api_key_parameter_name and credentials:

liquibase/changelog.xml

@@ -68,4 +68,5 @@
     <include file="changes/feat_1260.sql" relativeToChangelogFile="true"/>
     <include file="changes/feat_1333.sql" relativeToChangelogFile="true"/>
     <include file="changes/feat_pt_152.sql" relativeToChangelogFile="true"/>
+    <include file="changes/feat_1325.sql" relativeToChangelogFile="true"/>
 </databaseChangeLog>

liquibase/changes/feat_1325.sql

@@ -0,0 +1,25 @@
+ALTER TABLE feed ADD CONSTRAINT feed_stable_id_unique UNIQUE (stable_id);
+-- 1. Catalog of allowed keys + optional global values
+CREATE TABLE config_key (
+  namespace     text        NOT NULL,              -- e.g. 'reverse_geolocation'
+  key           text        NOT NULL,              -- e.g. 'country_fallback'
+  description   text,
+  default_value jsonb,                             -- fallback if nothing else
+  updated_at    timestamptz NOT NULL DEFAULT now(),
+  PRIMARY KEY (namespace, key)
+);
+
+-- 2. Per-feed overrides
+CREATE TABLE config_value_feed (
+  feed_stable_id       varchar(255) NOT NULL,
+  namespace     text         NOT NULL,
+  key           text         NOT NULL,
+  value         jsonb        NOT NULL,
+  updated_at    timestamptz  NOT NULL DEFAULT now(),
+  PRIMARY KEY (feed_stable_id, namespace, key),
+  FOREIGN KEY (feed_stable_id) REFERENCES feed(stable_id) ON DELETE CASCADE,
+  FOREIGN KEY (namespace, key) REFERENCES config_key(namespace, key) ON DELETE CASCADE
+);
+
+-- Helpful index
+CREATE INDEX config_value_feed_gin ON config_value_feed USING GIN (value);