add catalog config to manifest.py

core/dbt/config/external_config.py

@@ -1,6 +1,4 @@
-from typing import Optional
-
-from dbt_config.external_config import ExternalCatalogConfig
+from typing import Dict, Optional
 
 from dbt.clients.yaml_helper import load_yaml_text
 from dbt.constants import EXTERNAL_CATALOG_FILE_NAME
@@ -19,8 +17,8 @@ def _load_yml_dict(file_path):
     return None
 
 
-def load_external_catalog_config(project_root) -> Optional[ExternalCatalogConfig]:
-    unparsed_config = _load_yml_dict(f"{project_root}/{EXTERNAL_CATALOG_FILE_NAME}")
+def load_external_catalog_config(project) -> Optional[Dict]:
+    unparsed_config = _load_yml_dict(f"{project.project_root}/{EXTERNAL_CATALOG_FILE_NAME}")
     if unparsed_config is not None:
-        return ExternalCatalogConfig.model_validate(unparsed_config)
+        return unparsed_config
     return None

core/dbt/config/renderer.py

@@ -229,3 +229,9 @@ class PackageRenderer(SecretRenderer):
     @property
     def name(self):
         return "Packages config"
+
+
+class CatalogRenderer(SecretRenderer):
+    @property
+    def name(self):
+        return "Catalog config"

core/dbt/contracts/graph/manifest.py

@@ -20,6 +20,7 @@
     Union,
 )
 
+from dbt_config.external_config import ExternalCatalog
 from typing_extensions import Protocol
 
 import dbt_common.exceptions
@@ -844,6 +845,7 @@ class Manifest(MacroMethods, dbtClassMixin):
     unit_tests: MutableMapping[str, UnitTestDefinition] = field(default_factory=dict)
     saved_queries: MutableMapping[str, SavedQuery] = field(default_factory=dict)
     fixtures: MutableMapping[str, UnitTestFileFixture] = field(default_factory=dict)
+    catalogs: MutableMapping[str, ExternalCatalog] = field(default_factory=dict)
 
     _doc_lookup: Optional[DocLookup] = field(
         default=None, metadata={"serialize": lambda x: None, "deserialize": lambda x: None}

tests/functional/configs/fixtures.py

@@ -32,7 +32,7 @@
 
 models__untagged_sql = """
 {{
-    config(materialized='table')
+    config(materialized=table)
 }}
 
 select id, value from {{ source('raw', 'seed') }}

tests/unit/utils/__init__.py

@@ -6,12 +6,15 @@
 
 import os
 import string
+from typing import Dict
 from unittest import TestCase, mock
 
 import agate
 import pytest
+from dbt_config.external_config import ExternalCatalogConfig
 
 from dbt.config.project import PartialProject
+from dbt.config.renderer import CatalogRenderer
 from dbt.contracts.graph.manifest import Manifest
 from dbt_common.dataclass_schema import ValidationError
 
@@ -57,6 +60,14 @@ def profile_from_dict(profile, profile_name, cli_vars="{}"):
     )
 
 
+def catalog_from_dict(catalog, cli_vars=None):
+    if cli_vars is None:
+        cli_vars = {}
+    renderer = CatalogRenderer(cli_vars)
+    rendered = renderer.render_value(catalog)
+    return ExternalCatalogConfig.model_validate(rendered)
+
+
 def project_from_dict(project, profile, packages=None, selectors=None, cli_vars="{}"):
     from dbt.config.renderer import DbtProjectYamlRenderer
     from dbt.config.utils import parse_cli_vars
@@ -77,7 +88,9 @@ def project_from_dict(project, profile, packages=None, selectors=None, cli_vars=
     return partial.render(renderer)
 
 
-def config_from_parts_or_dicts(project, profile, packages=None, selectors=None, cli_vars={}):
+def config_from_parts_or_dicts(
+    project, profile, packages=None, selectors=None, cli_vars={}, catalogs=None
+):
     from copy import deepcopy
 
     from dbt.config import Profile, Project, RuntimeConfig
@@ -103,10 +116,13 @@ def config_from_parts_or_dicts(project, profile, packages=None, selectors=None,
             cli_vars,
         )
 
+    if isinstance(catalogs, Dict):
+        catalogs = catalog_from_dict(catalogs, cli_vars)
+
     args = Obj()
     args.vars = cli_vars
     args.profile_dir = "/dev/null"
-    return RuntimeConfig.from_parts(project=project, profile=profile, args=args)
+    return RuntimeConfig.from_parts(project=project, profile=profile, args=args, catalogs=catalogs)
 
 
 def inject_plugin(plugin):