Changed MergeStrategy from an Enum to constants.

cookie_composer/composition.py

@@ -3,7 +3,6 @@
 
 import logging
 import os
-from enum import Enum
 from pathlib import Path
 
 from pydantic import AnyHttpUrl, BaseModel, DirectoryPath, Field, root_validator
@@ -14,25 +13,24 @@
 logger = logging.getLogger(__name__)
 
 
-class MergeStrategy(str, Enum):
-    """Strategies of merging files and data."""
+# Strategies merging files and data.
 
-    DO_NOT_MERGE = "do-not-merge"
-    """Do not merge the data, use the file path to determine what to do."""
+DO_NOT_MERGE = "do-not-merge"
+"""Do not merge the data, use the file path to determine what to do."""
 
-    NESTED_OVERWRITE = "nested-overwrite"
-    """Merge deeply nested structures and overwrite at the lowest level; A deep ``dict.update()``."""
+NESTED_OVERWRITE = "nested-overwrite"
+"""Merge deeply nested structures and overwrite at the lowest level; A deep ``dict.update()``."""
 
-    OVERWRITE = "overwrite"
-    """Overwrite at the top level like ``dict.update()``."""
+OVERWRITE = "overwrite"
+"""Overwrite at the top level like ``dict.update()``."""
 
-    COMPREHENSIVE = "comprehensive"
-    """Comprehensively merge the two data structures.
+COMPREHENSIVE = "comprehensive"
+"""Comprehensively merge the two data structures.
 
-    - Scalars are overwritten by the new values
-    - lists are merged and de-duplicated
-    - dicts are recursively merged
-    """
+- Scalars are overwritten by the new values
+- lists are merged and de-duplicated
+- dicts are recursively merged
+"""
 
 
 class LayerConfig(BaseModel):
@@ -90,7 +88,7 @@ class LayerConfig(BaseModel):
     overwrite_exclude: List[str] = Field(default_factory=list)
     """Paths or glob patterns to exclude from overwriting."""
 
-    merge_strategies: Dict[str, MergeStrategy] = Field(default_factory=lambda: {"*": "do-not-merge"})
+    merge_strategies: Dict[str, str] = Field(default_factory=lambda: {"*": DO_NOT_MERGE})
     """The method to merge specific paths or glob patterns."""
 
 
@@ -195,11 +193,11 @@ def write_composition(layers: list, destination: Union[str, Path]):
         yaml.dump_all(dict_layers, f)
 
 
-def get_merge_strategy(path: Path, merge_strategies: Dict[str, str]) -> MergeStrategy:
+def get_merge_strategy(path: Path, merge_strategies: Dict[str, str]) -> str:
     """
     Return the merge strategy of the path based on the layer configured rules.
 
-    Files that are not mergable return ``MergeStrategy.DO_NOT_MERGE``
+    Files that are not mergable return ``DO_NOT_MERGE``
 
     Args:
         path: The file path to evaluate.
@@ -210,10 +208,10 @@ def get_merge_strategy(path: Path, merge_strategies: Dict[str, str]) -> MergeStr
     """
     from cookie_composer.merge_files import MERGE_FUNCTIONS
 
-    strategy = MergeStrategy.DO_NOT_MERGE  # The default
+    strategy = DO_NOT_MERGE  # The default
 
     if path.suffix not in MERGE_FUNCTIONS:
-        return MergeStrategy.DO_NOT_MERGE
+        return DO_NOT_MERGE
 
     for pattern, strat in merge_strategies.items():
         if rel_fnmatch(str(path), pattern):

cookie_composer/layers.py

@@ -1,5 +1,5 @@
 """Layer management."""
-from typing import Mapping
+from typing import List, Mapping, Optional
 
 import logging
 import os
@@ -9,12 +9,10 @@
 from pathlib import Path
 
 from cookie_composer.composition import (
+    DO_NOT_MERGE,
     LayerConfig,
-    MergeStrategy,
-    ProjectComposition,
     RenderedLayer,
     get_merge_strategy,
-    write_composition,
 )
 from cookie_composer.data_merge import comprehensive_merge
 from cookie_composer.matching import matches_any_glob
@@ -154,7 +152,7 @@ def get_write_strategy(origin: Path, destination: Path, rendered_layer: Rendered
         return WriteStrategy.WRITE
 
     merge_strat = get_merge_strategy(origin, rendered_layer.layer.merge_strategies)
-    if merge_strat != MergeStrategy.DO_NOT_MERGE:
+    if merge_strat != DO_NOT_MERGE:
         logger.debug("Strategy is not do-not-merge. Merging.")
         return WriteStrategy.MERGE
     logger.debug("Strategy is do-not-merge. Continuing evaluation.")
@@ -175,30 +173,35 @@ def get_write_strategy(origin: Path, destination: Path, rendered_layer: Rendered
         return WriteStrategy.WRITE
 
 
-def process_composition(composition: ProjectComposition):
-    """Process the composition."""
-    full_context = {}
+def render_layers(
+    layers: List[LayerConfig], destination: Path, initial_context: Optional[dict] = None, no_input: bool = False
+) -> List[RenderedLayer]:
+    """
+    Render layers to a destination.
+
+    Args:
+        layers: A list of ``LayerConfig`` to render
+        destination: The location to merge the rendered layers to
+        initial_context: An initial context to pass to the rendering
+        no_input: If ``True`` force each layer's ``no_input`` attribute to ``True``
+
+    Returns:
+        A list of the rendered layer information
+    """
+    full_context = initial_context or {}
     rendered_layers = []
 
-    for layer_config in composition.layers:
+    for layer_config in layers:
+        layer_config.no_input = True if no_input else layer_config.no_input
         if layer_config.context:
             full_context = comprehensive_merge(full_context, layer_config.context)
 
         with tempfile.TemporaryDirectory() as render_dir:
             rendered_layer = render_layer(layer_config, render_dir, full_context)
-            merge_layers(composition.destination, rendered_layer)
-        rendered_layers.append(rendered_layer)
-        full_context = comprehensive_merge(full_context, rendered_layer.new_context)
-
-    layers = []
-    layer_names = set()
-    for rendered_layer in rendered_layers:
+            merge_layers(destination, rendered_layer)
         rendered_layer.layer.commit = rendered_layer.latest_commit
         rendered_layer.layer.context = rendered_layer.new_context
-        layers.append(rendered_layer.layer)
-        layer_names.add(rendered_layer.layer_name)
+        rendered_layers.append(rendered_layer)
+        full_context = comprehensive_merge(full_context, rendered_layer.new_context)
 
-    if len(layer_names) != 1:
-        raise ValueError("Can not write composition file because multiple directories were rendered.")
-    composition_file = composition.destination / layer_names.pop() / ".composition.yaml"
-    write_composition(layers, composition_file)
+    return rendered_layers

cookie_composer/merge_files/__init__.py

@@ -26,11 +26,10 @@ def merge_generic_files(origin: Path, destination: Path, merge_strategy: MergeSt
 
 from pathlib import Path
 
-from cookie_composer.composition import MergeStrategy
 from cookie_composer.merge_files.json_file import merge_json_files
 from cookie_composer.merge_files.yaml_file import merge_yaml_files
 
-merge_function = Callable[[Path, Path, MergeStrategy], None]
+merge_function = Callable[[Path, Path, str], None]
 
 MERGE_FUNCTIONS: Dict[str, merge_function] = {
     ".json": merge_json_files,

cookie_composer/merge_files/json_file.py

@@ -3,13 +3,16 @@
 from pathlib import Path
 
 from cookie_composer import data_merge
-from cookie_composer.composition import MergeStrategy
+from cookie_composer.composition import (
+    COMPREHENSIVE,
+    DO_NOT_MERGE,
+    NESTED_OVERWRITE,
+    OVERWRITE,
+)
 from cookie_composer.exceptions import MergeError
 
 
-def merge_json_files(
-    new_file: Path, existing_file: Path, merge_strategy: MergeStrategy
-):
+def merge_json_files(new_file: Path, existing_file: Path, merge_strategy: str):
     """
     Merge two json files into one.
 
@@ -21,25 +24,25 @@ def merge_json_files(
     Raises:
         MergeError: If something goes wrong
     """
-    if merge_strategy == MergeStrategy.DO_NOT_MERGE:
+    if merge_strategy == DO_NOT_MERGE:
         raise MergeError(
             str(new_file),
             str(existing_file),
-            str(merge_strategy),
+            merge_strategy,
             "Can not merge with do-not-merge strategy.",
         )
 
     try:
         new_data = json.loads(new_file.read_text())
         existing_data = json.loads(existing_file.read_text())
     except (json.JSONDecodeError, FileNotFoundError) as e:
-        raise MergeError(str(new_file), str(existing_file), str(merge_strategy), str(e))
+        raise MergeError(str(new_file), str(existing_file), merge_strategy, str(e)) from e
 
-    if merge_strategy == MergeStrategy.OVERWRITE:
+    if merge_strategy == OVERWRITE:
         existing_data.update(new_data)
-    elif merge_strategy == MergeStrategy.NESTED_OVERWRITE:
+    elif merge_strategy == NESTED_OVERWRITE:
         existing_data = data_merge.deep_merge(existing_data, new_data)
-    elif merge_strategy == MergeStrategy.COMPREHENSIVE:
+    elif merge_strategy == COMPREHENSIVE:
         existing_data = data_merge.comprehensive_merge(existing_data, new_data)
     else:
         raise MergeError(error_message=f"Unrecognized merge strategy {merge_strategy}")

cookie_composer/merge_files/yaml_file.py

@@ -3,11 +3,16 @@
 from pathlib import Path
 
 from cookie_composer import data_merge
-from cookie_composer.composition import MergeStrategy
+from cookie_composer.composition import (
+    COMPREHENSIVE,
+    DO_NOT_MERGE,
+    NESTED_OVERWRITE,
+    OVERWRITE,
+)
 from cookie_composer.exceptions import MergeError
 
 
-def merge_yaml_files(new_file: Path, existing_file: Path, merge_strategy: MergeStrategy):
+def merge_yaml_files(new_file: Path, existing_file: Path, merge_strategy: str):
     """
     Merge two json files into one.
 
@@ -23,28 +28,28 @@ def merge_yaml_files(new_file: Path, existing_file: Path, merge_strategy: MergeS
 
     yaml = YAML(typ="safe")
 
-    if merge_strategy == MergeStrategy.DO_NOT_MERGE:
+    if merge_strategy == DO_NOT_MERGE:
         raise MergeError(
             str(new_file),
             str(existing_file),
-            str(merge_strategy),
+            merge_strategy,
             "Can not merge with do-not-merge strategy.",
         )
 
     try:
         new_data = yaml.load(new_file)
         existing_data = yaml.load(existing_file)
     except (YAMLError, FileNotFoundError) as e:
-        raise MergeError(str(new_file), str(existing_file), str(merge_strategy), str(e))
+        raise MergeError(str(new_file), str(existing_file), merge_strategy, str(e))
 
-    if merge_strategy == MergeStrategy.OVERWRITE:
+    if merge_strategy == OVERWRITE:
         if isinstance(existing_data, dict) and isinstance(new_data, dict):
             existing_data.update(new_data)
         else:
             existing_data = new_data
-    elif merge_strategy == MergeStrategy.NESTED_OVERWRITE:
+    elif merge_strategy == NESTED_OVERWRITE:
         existing_data = data_merge.deep_merge(existing_data, new_data)
-    elif merge_strategy == MergeStrategy.COMPREHENSIVE:
+    elif merge_strategy == COMPREHENSIVE:
         existing_data = data_merge.comprehensive_merge(existing_data, new_data)
     else:
         raise MergeError(error_message=f"Unrecognized merge strategy {merge_strategy}")