Provide utility functions for calling from hydra

ccflow/__init__.py

@@ -1,11 +1,16 @@
 __version__ = "0.6.10"
 
+# Import exttypes early so modules that import `from ccflow import PyObjectPath` during
+# initialization find it (avoids circular import issues with functions that import utilities
+# which, in turn, import `ccflow`).
+from .exttypes import *  # noqa: I001
+
 from .arrow import *
 from .base import *
+from .compose import *
 from .callable import *
 from .context import *
 from .enums import Enum
-from .exttypes import *
 from .global_state import *
 from .models import *
 from .object_config import *

ccflow/base.py

@@ -33,7 +33,6 @@
 log = logging.getLogger(__name__)
 
 __all__ = (
-    "model_alias",
     "BaseModel",
     "ModelRegistry",
     "ModelType",
@@ -323,20 +322,6 @@ def _is_config_subregistry(value):
     return False
 
 
-def model_alias(model_name: str) -> BaseModel:
-    """Function to alias a BaseModel by name in the root registry.
-
-    Useful for configs in hydra where we want a config object to point directly to another config object.
-
-    Args:
-        model_name: The name of the underlying model to point to in the registry
-    Example:
-        _target_: ccflow.model_alias
-        model_name: foo
-    """
-    return BaseModel.model_validate(model_name)
-
-
 ModelType = TypeVar("ModelType", bound=BaseModel)
 
 

ccflow/compose.py

@@ -0,0 +1,119 @@
+from typing import Any, Dict, Optional, Type, Union
+
+from .base import BaseModel
+from .exttypes.pyobjectpath import _TYPE_ADAPTER as PyObjectPathTA
+
+__all__ = (
+    "model_alias",
+    "from_python",
+    "update_from_template",
+)
+
+
+def model_alias(model_name: str) -> BaseModel:
+    """Return a model by alias from the registry.
+
+    Hydra-friendly: `_target_: ccflow.compose.model_alias` with `model_name`.
+
+    Args:
+        model_name: Alias string registered in the model registry. Typically a
+            short name that maps to a configured BaseModel.
+
+    Returns:
+        A ``BaseModel`` instance resolved from the registry by ``model_name``.
+    """
+    return BaseModel.model_validate(model_name)
+
+
+def from_python(py_object_path: str, indexer: Optional[list] = None) -> Any:
+    """Hydra-friendly: resolve and return any Python object by import path.
+
+    Optionally accepts ``indexer``, a list of keys that will be applied in
+    order to index into the resolved object. No safety checks are performed;
+    indexing errors will propagate.
+
+    Args:
+        py_object_path: Dotted import path to a Python object, e.g.
+            ``mypkg.module.OBJECT`` or ``mypkg.module.ClassName``.
+        indexer: Optional list of keys to apply in order to index into the
+            resolved object (e.g., strings for dict keys or integers for list
+            indexes).
+
+    Returns:
+        The resolved Python object, or the value obtained after applying all
+        ``indexer`` keys to the resolved object.
+
+    Example YAML usage:
+      some_value:
+        _target_: ccflow.compose.from_python
+        py_object_path: mypkg.module.OBJECT
+
+      nested_value:
+        _target_: ccflow.compose.from_python
+        py_object_path: mypkg.module.NESTED
+        indexer: ["a", "b"]
+    """
+    obj = PyObjectPathTA.validate_python(py_object_path).object
+    if indexer:
+        for key in indexer:
+            obj = obj[key]
+    return obj
+
+
+def update_from_template(
+    base: Optional[Union[str, Dict[str, Any], BaseModel]] = None,
+    *,
+    target_class: Optional[Union[str, Type]] = None,
+    update: Optional[Dict[str, Any]] = None,
+) -> Any:
+    """Generic update helper that constructs an instance from a base and updates.
+
+    Args:
+        base: Either a registry alias string, a dict, or a Pydantic BaseModel. If BaseModel, it is converted
+              to a shallow dict via ``dict(base)`` to preserve nested object identity.
+        target_class: Optional path to the target class to construct. May be a
+              string import path or the type itself. If None and ``base`` is a
+              BaseModel, returns an instance of ``base.__class__``. If None and
+              ``base`` is a dict, returns the updated dict.
+        update: Optional dict of updates to apply.
+
+    Returns:
+        Instance of ``target_class`` if provided; otherwise an instance of the same
+        class as ``base`` when base is a BaseModel; or the updated dict when base
+        is a dict.
+    """
+    # Determine base dict and default target
+    default_target = None
+    if isinstance(base, str):
+        # Allow passing alias name directly; resolve from registry
+        base = model_alias(base)
+    if isinstance(base, BaseModel):
+        base_dict = dict(base)
+        default_target = base.__class__
+    elif isinstance(base, dict):
+        base_dict = dict(base)
+    elif base is None:
+        base_dict = {}
+    else:
+        raise TypeError("base must be a dict, BaseModel, or None")
+
+    # Merge updates: explicit dict first, then kwargs
+    if update:
+        base_dict.update(update)
+
+    # Resolve target class if provided as string path
+    target = None
+    if target_class is not None:
+        if isinstance(target_class, str):
+            target = PyObjectPathTA.validate_python(target_class).object
+        else:
+            target = target_class
+    else:
+        target = default_target
+
+    if target is None:
+        # No target: return dict update for dict base
+        return base_dict
+
+    # Construct instance of target with updated fields
+    return target(**base_dict)

ccflow/tests/config/conf_from_python.yaml

@@ -0,0 +1,23 @@
+shared_model:
+  _target_: ccflow.compose.from_python
+  py_object_path: ccflow.tests.data.python_object_samples.SHARED_MODEL
+
+consumer:
+  _target_: ccflow.tests.data.python_object_samples.Consumer
+  shared: shared_model
+  tag: consumer1
+
+# Demonstrate from_python returning a dict (non-BaseModel)
+holder:
+  _target_: ccflow.tests.data.python_object_samples.SharedHolder
+  name: holder1
+  cfg:
+    _target_: ccflow.compose.from_python
+    py_object_path: ccflow.tests.data.python_object_samples.SHARED_CFG
+
+# Use update_from_template to update a field while preserving shared identity
+consumer_updated:
+  _target_: ccflow.compose.update_from_template
+  base: consumer
+  update:
+    tag: consumer2

ccflow/tests/data/__init__.py

@@ -0,0 +1,4 @@
+"""Shared test data modules.
+
+Import sample configs using module-level objects in `python_object_samples`.
+"""

ccflow/tests/data/python_object_samples.py

@@ -0,0 +1,33 @@
+"""Sample python objects for testing from_python and identity preservation."""
+
+from typing import Dict
+
+from ccflow import BaseModel
+
+# Module-level objects
+SHARED_CFG: Dict[str, int] = {"x": 1, "y": 2}
+OTHER_CFG: Dict[str, int] = {"x": 10, "y": 20}
+"""Dict samples; identity for dicts is not guaranteed by Pydantic."""
+
+NESTED_CFG = {
+    "db": {"host": "seed.local", "port": 7000, "name": "seed"},
+    "meta": {"env": "dev"},
+}
+
+
+class SharedHolder(BaseModel):
+    name: str
+    cfg: Dict[str, int]
+
+
+class SharedModel(BaseModel):
+    val: int = 0
+
+
+# Module-level instance to be resolved via from_python
+SHARED_MODEL = SharedModel(val=42)
+
+
+class Consumer(BaseModel):
+    shared: SharedModel
+    tag: str = ""