Dcv 2973 dbt coves implement datacoves secrets

dbt_coves/config/config.py

@@ -66,7 +66,7 @@ class GenerateAirflowDagsModel(BaseModel):
     secrets_manager: Optional[str] = ""
     secrets_url: Optional[str] = ""
     secrets_token: Optional[str] = ""
-    secrets_project: Optional[str] = ""
+    secrets_environment: Optional[str] = ""
     secrets_tags: Optional[str] = ""
     secrets_key: Optional[str] = ""
 
@@ -100,7 +100,7 @@ class LoadAirbyteModel(BaseModel):
     secrets_manager: Optional[str] = ""
     secrets_url: Optional[str] = ""
     secrets_token: Optional[str] = ""
-    secrets_project: Optional[str] = ""
+    secrets_environment: Optional[str] = ""
     secrets_tags: Optional[List[str]] = []
     secrets_key: Optional[str] = ""
 
@@ -114,7 +114,7 @@ class LoadFivetranModel(BaseModel):
     secrets_manager: Optional[str] = ""
     secrets_url: Optional[str] = ""
     secrets_token: Optional[str] = ""
-    secrets_project: Optional[str] = ""
+    secrets_environment: Optional[str] = ""
     secrets_tags: Optional[List[str]] = []
     secrets_key: Optional[str] = ""
 
@@ -224,7 +224,7 @@ class DbtCovesConfig:
         "generate.airflow_dags.secrets_manager",
         "generate.airflow_dags.secrets_url",
         "generate.airflow_dags.secrets_token",
-        "generate.airflow_dags.secrets_project",
+        "generate.airflow_dags.secrets_environment",
         "generate.airflow_dags.secrets_tags",
         "generate.airflow_dags.secrets_key",
         "extract.airbyte.path",
@@ -237,7 +237,7 @@ class DbtCovesConfig:
         "load.airbyte.secrets_manager",
         "load.airbyte.secrets_url",
         "load.airbyte.secrets_token",
-        "load.airbyte.secrets_project",
+        "load.airbyte.secrets_environment",
         "load.airbyte.secrets_tags",
         "load.airbyte.secrets_key",
         "setup.no_prompt",
@@ -259,7 +259,7 @@ class DbtCovesConfig:
         "load.fivetran.secrets_manager",
         "load.fivetran.secrets_url",
         "load.fivetran.secrets_token",
-        "load.fivetran.secrets_project",
+        "load.fivetran.secrets_environment",
         "load.fivetran.secrets_tags",
         "load.fivetran.secrets_key",
         "data_sync.redshift.tables",

dbt_coves/tasks/generate/airflow_dags.py

@@ -13,7 +13,7 @@
 
 from dbt_coves.core.exceptions import MissingArgumentException
 from dbt_coves.tasks.base import NonDbtBaseTask
-from dbt_coves.utils.secrets import load_secret_manager_data
+from dbt_coves.utils.secrets import load_secret_manager_data, replace_secrets
 from dbt_coves.utils.tracking import trackable
 from dbt_coves.utils.yaml import deep_merge
 
@@ -94,7 +94,7 @@ def register_parser(cls, sub_parsers, base_subparser):
         subparser.add_argument(
             "--secrets-token", type=str, help="Secret credentials provider token"
         )
-        subparser.add_argument("--secrets-project", type=str, help="Secret credentials project")
+        subparser.add_argument("--secrets-environment", type=str, help="Secret credentials project")
         subparser.add_argument("--secrets-tags", type=str, help="Secret credentials tags")
         subparser.add_argument("--secrets-key", type=str, help="Secret credentials key")
 
@@ -267,10 +267,13 @@ def build_dag_file(self, destination_path: Path, dag_name: str, yml_dag: Dict[st
                 console.print(f"DAG {dag_name} resulted in an invalid DAG, skipping. Error: {exc}")
 
     def _merge_secret_nodes(self, secret_nodes, yml_dag) -> Dict[str, Any]:
-        for node_name, node_config in secret_nodes.get("nodes", {}).items():
-            yml_node = yml_dag.get("nodes", {}).get(node_name)
-            if yml_node:
-                yml_dag["nodes"][node_name] = deep_merge(node_config, yml_node)
+        if isinstance(secret_nodes, dict):
+            for node_name, node_config in secret_nodes.get("nodes", {}).items():
+                yml_node = yml_dag.get("nodes", {}).get(node_name)
+                if yml_node:
+                    yml_dag["nodes"][node_name] = deep_merge(node_config, yml_node)
+        elif isinstance(secret_nodes, list):  # Datacoves secrets
+            replace_secrets(secret_nodes, yml_dag)
         return yml_dag
 
     def _discover_secrets(self, yml_dag: Dict[str, Any]):
@@ -283,9 +286,9 @@ def _discover_secrets(self, yml_dag: Dict[str, Any]):
                 yml_dag = self._merge_secret_nodes(secret_data, yml_dag)
 
         if self.secrets_manager:
-            secret_data = load_secret_manager_data(self)
-            for secret in secret_data:
-                yml_dag = self._merge_secret_nodes(secret.get("value", {}), yml_dag)
+            self.secret_data = load_secret_manager_data(self)
+            if self.secret_data:
+                yml_dag = self._merge_secret_nodes(self.secret_data, yml_dag)
 
         return yml_dag
 
@@ -321,6 +324,8 @@ def _merge_generator_configs(self, tg_conf: Dict[str, Any], generator: str) -> D
         """
         generators_params = self.get_config_value("generators_params")
         coves_config_generators_params = generators_params.get(generator, {})
+        if self.secrets_manager:
+            replace_secrets(self.secret_data, coves_config_generators_params)
         return deep_merge(tg_conf, coves_config_generators_params)
 
     def generate_task_group(self, tg_name: str, tg_conf: Dict[str, Any]):

dbt_coves/tasks/load/airbyte.py

@@ -63,7 +63,7 @@ def register_parser(cls, sub_parsers, base_subparser):
         subparser.add_argument(
             "--secrets-token", type=str, help="Secret credentials provider token"
         )
-        subparser.add_argument("--secrets-project", type=str, help="Secret credentials project")
+        subparser.add_argument("--secrets-environment", type=str, help="Secret credentials project")
         subparser.add_argument("--secrets-tags", type=str, help="Secret credentials tags")
         subparser.add_argument("--secrets-key", type=str, help="Secret credentials key")
         subparser.set_defaults(cls=cls, which="airbyte")

dbt_coves/tasks/load/fivetran.py

@@ -57,7 +57,7 @@ def register_parser(cls, sub_parsers, base_subparser):
         subparser.add_argument(
             "--secrets-token", type=str, help="Secret credentials provider token"
         )
-        subparser.add_argument("--secrets-project", type=str, help="Secret credentials project")
+        subparser.add_argument("--secrets-environment", type=str, help="Secret credentials project")
         subparser.add_argument("--secrets-tags", type=str, help="Secret credentials tags")
         subparser.add_argument("--secrets-key", type=str, help="Secret credentials key")
         subparser.set_defaults(cls=cls, which="fivetran")

dbt_coves/utils/flags.py

@@ -84,7 +84,7 @@ def __init__(self, cli_parser: ArgumentParser) -> None:
                 "secrets_manager": None,
                 "secrets_url": None,
                 "secrets_token": None,
-                "secrets_project": None,
+                "secrets_environment": None,
                 "secrets_tags": None,
                 "secrets_key": None,
             },
@@ -111,7 +111,7 @@ def __init__(self, cli_parser: ArgumentParser) -> None:
                 "secrets_manager": None,
                 "secrets_url": None,
                 "secrets_token": None,
-                "secrets_project": None,
+                "secrets_environment": None,
                 "secrets_tags": None,
                 "secrets_key": None,
             },
@@ -124,7 +124,7 @@ def __init__(self, cli_parser: ArgumentParser) -> None:
                 "secrets_manager": None,
                 "secrets_url": None,
                 "secrets_token": None,
-                "secrets_project": None,
+                "secrets_environment": None,
                 "secrets_tags": None,
                 "secrets_key": None,
             },
@@ -309,8 +309,10 @@ def parse_args(self, cli_args: List[str] = list()) -> None:
                     self.generate["airflow_dags"]["secrets_url"] = self.args.secrets_url
                 if self.args.secrets_token:
                     self.generate["airflow_dags"]["secrets_token"] = self.args.secrets_token
-                if self.args.secrets_project:
-                    self.generate["airflow_dags"]["secrets_project"] = self.args.secrets_project
+                if self.args.secrets_environment:
+                    self.generate["airflow_dags"][
+                        "secrets_environment"
+                    ] = self.args.secrets_environment
                 if self.args.secrets_tags:
                     self.generate["airflow_dags"]["secrets_tags"] = self.args.secrets_tags
                 if self.args.secrets_key:
@@ -332,8 +334,8 @@ def parse_args(self, cli_args: List[str] = list()) -> None:
                     self.load["airbyte"]["secrets_url"] = self.args.secrets_url
                 if self.args.secrets_token:
                     self.load["airbyte"]["secrets_token"] = self.args.secrets_token
-                if self.args.secrets_project:
-                    self.load["airbyte"]["secrets_project"] = self.args.secrets_project
+                if self.args.secrets_environment:
+                    self.load["airbyte"]["secrets_environment"] = self.args.secrets_environment
                 if self.args.secrets_tags:
                     self.load["airbyte"]["secrets_tags"] = [
                         tag.strip() for tag in self.args.secrets_tags.split(",")
@@ -359,8 +361,8 @@ def parse_args(self, cli_args: List[str] = list()) -> None:
                     self.load["fivetran"]["secrets_url"] = self.args.secrets_url
                 if self.args.secrets_token:
                     self.load["fivetran"]["secrets_token"] = self.args.secrets_token
-                if self.args.secrets_project:
-                    self.load["fivetran"]["secrets_project"] = self.args.secrets_project
+                if self.args.secrets_environment:
+                    self.load["fivetran"]["secrets_environment"] = self.args.secrets_environment
                 if self.args.secrets_tags:
                     self.load["fivetran"]["secrets_tags"] = [
                         tag.strip() for tag in self.args.secrets_tags.split(",")

dbt_coves/utils/secrets.py

@@ -1,9 +1,12 @@
 import os
+import re
 
 import requests
 
 from dbt_coves.core.exceptions import DbtCovesException
 
+SECRET_PATTERN = re.compile(r"\{\{\s*secret\('([^']+)'\)\s*\}\}", re.IGNORECASE)
+
 
 def load_secret_manager_data(task_instance) -> dict:
     payload = {}
@@ -16,17 +19,16 @@ def load_secret_manager_data(task_instance) -> dict:
         secrets_token = os.getenv("DATACOVES__SECRETS_TOKEN") or task_instance.get_config_value(
             "secrets_token"
         )
-        secrets_project = os.getenv("DATACOVES__SECRETS_PROJECT") or task_instance.get_config_value(
-            "secrets_project"
-        )
-
-        if not (secrets_url and secrets_token and secrets_project):
+        secrets_environment = os.getenv(
+            "DATACOVES__ENVIRONMENT_SLUG"
+        ) or task_instance.get_config_value("secrets_environment")
+        if not (secrets_url and secrets_token and secrets_environment):
             raise DbtCovesException(
-                "[b]secrets_url[/b], [b]secrets_project[/b] and [b]secrets_token[/b] must "
+                "[b]secrets_url[/b], [b]secrets_environment[/b] and [b]secrets_environment[/b] must "
                 "be provided when using a Secrets Manager"
             )
 
-        secrets_url = f"{secrets_url}/api/v1/secrets/{secrets_project}"
+        secrets_url = f"{secrets_url}/api/v1/secrets/{secrets_environment}"
         secrets_tags = task_instance.get_config_value("secrets_tags")
         secrets_key = task_instance.get_config_value("secrets_key")
         if secrets_tags:
@@ -42,3 +44,30 @@ def load_secret_manager_data(task_instance) -> dict:
         return response.json()
 
     raise DbtCovesException(f"'{manager}' not recognized as a valid secrets manager.")
+
+
+def replace_secrets(secrets_list, dictionary, errors: set = set()):
+    for key, value in dictionary.items():
+        if isinstance(value, dict):
+            replace_secrets(secrets_list, value, errors)
+        elif isinstance(value, str):
+            value_secret = SECRET_PATTERN.search(value)
+            if value_secret:
+                secret_key = value_secret.group(1)
+                secret_found = False
+                for secret in secrets_list:
+                    if secret.get("slug", "").lower() == secret_key.lower():
+                        secret_found = True
+                        if secret.get("slug", "") == secret_key:
+                            dictionary[key] = secret.get("value")
+                        else:
+                            errors.add(
+                                f"Secret [red]{secret_key}[/red] not found in secrets, "
+                                f"did you mean [red]{secret.get('slug')}[/red]"
+                            )
+                if not secret_found:
+                    errors.add(f"Secret [red]{secret_key}[/red] not found in secrets")
+    if errors:
+        error_message = "Errors found:\n"
+        error_message += "\n".join(errors)
+        raise DbtCovesException(error_message)