Rudimentary catalog init refactoring to use a config file

Author: rnapark

deltacat/catalog/model/catalog.py

@@ -12,6 +12,7 @@
 from deltacat.catalog.main import impl as dcat
 from deltacat.catalog.model.properties import CatalogProperties
 from deltacat.constants import DEFAULT_CATALOG
+from deltacat.utils.config_loader import load_catalog_configs_from_yaml
 
 all_catalogs: Optional[ray.actor.ActorHandle] = None
 
@@ -182,6 +183,7 @@ def init(
     ray_init_args: Dict[str, Any] = {},
     *,
     force=False,
+    config_path: Optional[str] = None,
 ) -> Optional[ray.runtime.BaseContext]:
     """
     Initialize DeltaCAT catalogs.
@@ -202,6 +204,16 @@ def init(
         logger.warning("DeltaCAT already initialized.")
         return None
 
+    # If catalogs are provided and a config_path is also provided, raise ValueError
+    if catalogs and config_path is not None:
+        raise ValueError(
+            "Cannot provide both `catalogs` and `config_path`. Please provide "
+            "only one of these parameters."
+        )
+    # If no catalogs provided but a config_path exists, create a Catalog
+    if not catalogs and config_path is not None:
+        catalogs = load_catalog_configs_from_yaml(config_path=config_path)
+
     # initialize ray (and ignore reinitialization errors)
     ray_init_args["ignore_reinit_error"] = True
     context = ray.init(**ray_init_args)

deltacat/tests/catalog/test_catalogs.py

@@ -4,6 +4,7 @@
 import uuid
 from unittest import mock
 import os
+import yaml
 
 from deltacat.catalog import (
     CatalogProperties,
@@ -147,11 +148,54 @@ def test_init_with_default_catalog_name(self, reset_catalogs):
             force=True,
         )
 
-        # Get the default catalog and check it's catalog2
+        # Get the def ault catalog and check it's catalog2
         default_catalog = get_catalog()
         assert default_catalog.impl == MockCatalogImpl
         assert default_catalog.inner["kwargs"]["id"] == 2
 
+    def test_init_with_config_yaml(self, tmp_path, reset_catalogs):
+        """
+        Test initializing Catalogs from a YAML config file.
+        CatalogProperties will infer filesystem and storage, so we don't
+        put actual Python objects in the YAML.
+        - load_catalog_configs_from_yaml returns a dict[str, CatalogProperties],
+        so we provide one entry keyed by a catalog name.
+        - get_catalog() with no arguments should always return the *default*
+        catalog, even if multiple catalogs are loaded. In this case we only
+        provide one catalog ("test_catalog"), so it should automatically
+        become the default.
+        """
+        # YAML data compatible with CatalogProperties
+        config_data = {
+            "test-catalog": {
+                "root": str(tmp_path),  # path for catalog metadata/data
+                "filesystem": None,  # leave None; CatalogProperties will infer filesystem
+                "storage": None,  # leave None; CatalogProperties will infer storage
+            }
+        }
+
+        # Write the YAML config file
+        config_path = tmp_path / "config.yaml"
+        with open(config_path, "w") as f:
+            yaml.dump(config_data, f)
+
+        # Initialize from the YAML config
+        init(config_path=str(config_path), force=True)
+
+        # Retrieve the default catalog (returns CatalogProperties directly)
+        catalog_props = get_catalog()  # no args → default "test-catalog"
+
+        # Should be a CatalogProperties instance
+        assert isinstance(catalog_props, CatalogProperties)
+        assert catalog_props.root == str(tmp_path)
+
+        # filesystem and storage are inferred by CatalogProperties; check types
+        import pyarrow.fs
+
+        assert isinstance(catalog_props.filesystem, pyarrow.fs.FileSystem)
+        # TODO: If storage has a default class, check its type here
+        # assert isinstance(inner.storage, ExpectedStorageClass)
+
     def test_put_catalog(self, reset_catalogs):
         """Test adding a catalog after initialization."""
         # Initialize with a single catalog

deltacat/utils/config_loader.py

@@ -0,0 +1,46 @@
+import yaml
+from typing import Dict
+from deltacat.catalog.model.properties import CatalogProperties
+
+
+def load_catalog_configs_from_yaml(config_path: str) -> Dict[str, CatalogProperties]:
+    """
+    Load one or more catalog configs from a YAML file.
+
+    The YAML can either be:
+      1. A single unnamed config (a dict of properties) -> wrapped as {"default": CatalogProperties}
+      2. A dictionary of named configs: name -> property-mapping
+
+    Args:
+        config_path: Path to the YAML config file.
+
+    Returns:
+        Dict[str, CatalogProperties]: Mapping of Catalog name -> CatalogProperties.
+    """
+    with open(config_path, "r") as f:
+        config_data = yaml.safe_load(f)
+
+    if not isinstance(config_data, dict):
+        raise ValueError(
+            f"Invalid YAML format in {config_path}. "
+            f"Expected a dict, got {type(config_data)}"
+        )
+
+    # Case 1: single unnamed config
+    # e.g. {"type": "iceberg", "uri": "...", "warehouse": "prod"}
+    if all(
+        isinstance(v, (str, int, float, bool, type(None))) or isinstance(v, list)
+        for v in config_data.values()
+    ):
+        return {"default": CatalogProperties(**config_data)}
+
+    # Case 2: top-level dict of name -> dict-of-properties
+    catalogs: Dict[str, CatalogProperties] = {}
+    for name, props in config_data.items():
+        if not isinstance(props, dict):
+            raise ValueError(
+                f"Config for catalog '{name}' must be a mapping, got {type(props)}"
+            )
+        catalogs[name] = CatalogProperties(**props)
+
+    return catalogs