Added code

cookie_composer/__init__.py

@@ -0,0 +1,2 @@
+"""The Cookiecutter Composer."""
+__version__: str = "0.1.0"

cookie_composer/_commands/_create.py

@@ -0,0 +1,37 @@
+"""Methods for generating projects."""
+
+from typing import Optional
+
+from pathlib import Path
+
+from cookie_composer.composition import (
+    LayerConfig,
+    ProjectComposition,
+    is_composition_file,
+    read_composition,
+)
+from cookie_composer.layers import process_composition
+
+
+def create(
+    path_or_url: str,
+    output_dir: Optional[Path] = None,
+) -> Path:
+    """
+    Generate a new project from a composition file, local template or remote template.
+
+    Args:
+        path_or_url: The path or url to the composition file or template
+        output_dir: Where to generate the project
+
+    Returns:
+        The path to the generated project.
+    """
+    output_dir = output_dir or Path(".")
+    if is_composition_file(path_or_url):
+        composition = read_composition(path_or_url, output_dir)
+    else:
+        tmpl = LayerConfig(template=path_or_url)
+        composition = ProjectComposition(layers=[tmpl], destination=output_dir)
+    process_composition(composition)
+    return output_dir

cookie_composer/cli.py

@@ -0,0 +1,39 @@
+"""Command line setup."""
+from typing import Optional
+
+from pathlib import Path
+
+import typer
+
+from cookie_composer._commands import _create
+
+app = typer.Typer()
+
+
+@app.command()
+def create(path_or_url: str, output_dir: Optional[Path] = None):
+    """
+    Create a project from a template or configuration.
+
+    Args:
+        path_or_url: The path or URL to the template or composition file
+        output_dir: Where to write the output
+    """
+    _create.create(path_or_url, output_dir)
+
+
+@app.command()
+def add(path_or_url: str):
+    """
+    Add a template or configuration to an existing project.
+
+    Args:
+        path_or_url: A URL or string to add the template or configuration
+    """
+
+
+@app.command()
+def update():
+    """
+    Update the project to the latest version of each template.
+    """

cookie_composer/composition.py

@@ -0,0 +1,210 @@
+"""Project configuration and options."""
+from typing import Any, Dict, List, Optional, Union
+
+import logging
+from enum import Enum
+from pathlib import Path
+
+from pydantic import AnyHttpUrl, BaseModel, DirectoryPath, Field
+
+from cookie_composer.exceptions import MissingCompositionFileError
+from cookie_composer.matching import rel_fnmatch
+
+logger = logging.getLogger(__name__)
+
+
+class MergeStrategy(str, Enum):
+    """Strategies of merging files and data."""
+
+    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()``."""
+
+    OVERWRITE = "overwrite"
+    """Overwrite at the top level like ``dict.update()``."""
+
+    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
+    """
+
+
+class LayerConfig(BaseModel):
+    """Configuration for a layer of a composition."""
+
+    #
+    # Template specification
+    #
+    template: Union[str, AnyHttpUrl]
+    """The path or URL to the template."""
+
+    directory: Optional[str]
+    """Directory within a git repository template that holds the cookiecutter.json file."""
+
+    checkout: Optional[str]
+    """The branch, tag or commit to use if template is a git repository.
+
+    Also used for updating projects."""
+
+    password: Optional[str]
+    """The password to use if template is a password-protected Zip archive."""
+
+    commit: Optional[str]
+    """What git hash was applied if the template is a git repository."""
+
+    #
+    # Input specification
+    #
+    no_input: bool = False
+    """Do not prompt for parameters and only use cookiecutter.json file content.
+
+    This is only used for initial generation. After initial generation, the results
+    are stored in the context."""
+
+    context: Dict[str, Any] = Field(default_factory=dict)
+    """Dictionary that will provide values for input.
+
+    Also stores the answers for missing context parameters after initial generation."""
+
+    #
+    # File generation
+    #
+    skip_hooks: bool = False
+    """Skip the template hooks."""
+
+    skip_if_file_exists: bool = True
+    """Skip the files in the corresponding directories if they already exist."""
+
+    skip_generation: List[str] = Field(default_factory=list)
+    """Paths or glob patterns to skip attempting to generate."""
+
+    overwrite: List[str] = Field(default_factory=list)
+    """Paths or glob patterns to always overwrite."""
+
+    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"}
+    )
+    """The method to merge specific paths or glob patterns."""
+
+
+class RenderedLayer(BaseModel):
+    """Information about a rendered layer."""
+
+    layer: LayerConfig
+    """The original layer configuration that was rendered."""
+
+    location: DirectoryPath
+    """The location to the rendered layer."""
+
+    new_context: Dict[str, Any]
+    """The context based on questions asked."""
+
+    latest_commit: Optional[str] = None
+    """The latest commit checkout out."""
+
+
+class ProjectComposition(BaseModel):
+    """Composition of templates for a project."""
+
+    layers: List[LayerConfig]
+    destination: DirectoryPath
+
+
+def is_composition_file(path_or_url: Union[str, Path]) -> bool:
+    """
+    Is the filename a composition file?
+
+    Args:
+        path_or_url: The path or URL to check
+
+    Returns:
+        ``True`` if the path is a configuration file.
+    """
+    return Path(path_or_url).suffix in {".yaml", ".yml"}
+
+
+def read_composition(
+    path_or_url: Union[str, Path], destination: Union[str, Path]
+) -> ProjectComposition:
+    """
+    Read a JSON or YAML file and return a ProjectComposition.
+
+    Args:
+        path_or_url: The location of the configuration file
+        destination: Where the destination of the project should be rendered
+
+    Returns:
+        A project composition
+
+    Raises:
+        MissingCompositionFileError: Raised when it can not access the configuration file.
+    """
+    import fsspec
+    from ruyaml import YAML
+
+    yaml = YAML(typ="safe")
+    try:
+        of = fsspec.open(path_or_url, mode="rt")
+        with of as f:
+            contents = list(yaml.load_all(f))
+            templates = [LayerConfig(**doc) for doc in contents]
+        return ProjectComposition(
+            layers=templates, destination=Path(destination).expanduser().resolve()
+        )
+    except (ValueError, FileNotFoundError) as e:
+        raise MissingCompositionFileError(path_or_url) from e
+
+
+def write_composition(layers: list, destination: Union[str, Path]):
+    """
+    Write a JSON or YAML composition file.
+
+    Args:
+        layers: The layers of the composition
+        destination: Where to write the file
+    """
+    import fsspec
+    from ruyaml import YAML
+
+    yaml = YAML(typ="safe")
+    of = fsspec.open(destination, mode="wt")
+    dict_layers = [layer.dict() for layer in layers]
+    with of as f:
+        yaml.dump_all(dict_layers, f)
+
+
+def get_merge_strategy(path: Path, merge_strategies: Dict[str, str]) -> MergeStrategy:
+    """
+    Return the merge strategy of the path based on the layer configured rules.
+
+    Files that are not mergable return ``MergeStrategy.DO_NOT_MERGE``
+
+    Args:
+        path: The file path to evaluate.
+        merge_strategies: The glob pattern->strategy mapping
+
+    Returns:
+        The appropriate merge strategy.
+    """
+    from cookie_composer.merge_files import MERGE_FUNCTIONS
+
+    strategy = MergeStrategy.DO_NOT_MERGE  # The default
+
+    if path.suffix not in MERGE_FUNCTIONS:
+        return MergeStrategy.DO_NOT_MERGE
+
+    for pattern, strat in merge_strategies.items():
+        if rel_fnmatch(str(path), pattern):
+            logger.debug(f"{path} matches merge strategy pattern {pattern} for {strat}")
+            strategy = strat
+            break
+
+    return strategy

cookie_composer/data_merge.py

@@ -0,0 +1,97 @@
+"""Tools for merging data."""
+from typing import Any, Iterable
+
+import copy
+from functools import reduce
+
+
+def deep_merge(*dicts) -> dict:
+    """
+    Merges dicts deeply.
+
+    Args:
+        dicts: List of dicts to merge with the first one the base
+
+    Returns:
+        dict: The merged dict
+    """
+
+    def merge_into(d1, d2):
+        for key in d2:
+            if key not in d1 or not isinstance(d1[key], dict):
+                d1[key] = copy.deepcopy(d2[key])
+            else:
+                d1[key] = merge_into(d1[key], d2[key])
+        return d1
+
+    return reduce(merge_into, dicts, {})
+
+
+def merge_iterables(iter1: Iterable, iter2: Iterable) -> set:
+    """
+    Merge and de-duplicate a bunch of lists into a single list.
+
+    Order is not guaranteed.
+
+    Args:
+        iter1: An Iterable
+        iter2: An Iterable
+
+    Returns:
+        The merged, de-duplicated sequence as a set
+    """
+    from itertools import chain
+
+    return set(chain(iter1, iter2))
+
+
+def comprehensive_merge(*args) -> Any:
+    """
+    Merges data comprehensively.
+
+    All arguments must be of the same type.
+
+    - Scalars are overwritten by the new values
+    - lists are merged and de-duplicated
+    - dicts are recursively merged
+
+    Args:
+        args: List of dicts to merge with the first one the base
+
+    Returns:
+        The merged data
+
+    Raises:
+        ValueError: If the values are not of the same type
+    """
+
+    def merge_into(d1, d2):
+        if type(d1) != type(d2):
+            raise ValueError(f"Cannot merge {type(d2)} into {type(d1)}.")
+
+        if isinstance(d1, list):
+            return list(merge_iterables(d1, d2))
+        elif isinstance(d1, set):
+            return merge_iterables(d1, d2)
+        elif isinstance(d1, tuple):
+            return tuple(merge_iterables(d1, d2))
+        elif isinstance(d1, dict):
+            for key in d2:
+                if key in d1:
+                    d1[key] = merge_into(d1[key], d2[key])
+                else:
+                    d1[key] = copy.deepcopy(d2[key])
+            return d1
+        else:
+            return copy.deepcopy(d2)
+
+    if isinstance(args[0], list):
+        return reduce(merge_into, args, [])
+    elif isinstance(args[0], tuple):
+        return reduce(merge_into, args, tuple())
+    elif isinstance(args[0], set):
+        return reduce(merge_into, args, set())
+    elif isinstance(args[0], dict):
+        return reduce(merge_into, args, {})
+    else:
+        return reduce(merge_into, args)