> ## Documentation Index
> Fetch the complete documentation index at: https://astronomer.io/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Merge yaml configurations

When merging YAML configurations into `values.yaml`, you can merge manually or with a tool of your choosing.

You can use the following `merge_yaml.py` script to merge YAML excerpts into `values.yaml` automatically. This script requires both Python and the `ruamel.yaml` package, which you can install using `pip install ruamel.yaml`.
To run the program, ensure that `merge_yaml.py`, `values.yaml`, and the `yaml` file that contains the configuration you want to add are all in your project directory. Then, run:

```sh wrap theme={null}
python merge_yaml.py values-to-merge.yaml values.yaml
```

```python expandable wrap theme={null}
#!/usr/bin/env python
"""
Backup destination file and merge YAML contents of src into dest.

By default creates backups, overwrites destination, and clobbers lists.

Usage:
    merge_yaml.py src dest [--create-backup=True] [--dry-run] [--show-stacktrace=False] [--merge-lists=True] [--help]
"""

import argparse
import os
import shutil
from datetime import datetime
import sys
from pathlib import Path

# Check Python version
if sys.version_info < (3, 0):
    print("Error: This script requires Python 3.0 or greater.")
    sys.exit(2)

# Try importing ruamel.yaml
try:
    from ruamel.yaml import YAML
except ImportError:
    print(
        "Error: ruamel.yaml is not installed. Please install it using 'pip install ruamel.yaml'"
    )
    sys.exit(2)

yaml = YAML()

def deep_merge(d1, d2, **kwargs):
    """Deep merges dictionary d2 into dictionary d1."""
    merge_lists = kwargs.get("merge_lists")
    for key, value in d2.items():
        if key in d1:
            if isinstance(d1[key], dict) and isinstance(value, dict):
                deep_merge(d1[key], value, **kwargs)
            elif merge_lists and isinstance(d1[key], list) and isinstance(value, list):
                d1[key].extend(value)
            else:
                d1[key] = value
        else:
            d1[key] = value
    return d1

def load_yaml_file(filename):
    """Load YAML data from a file."""
    if not os.path.exists(filename):
        return {}
    with open(filename, "r") as file:
        return yaml.load(file)

def save_yaml_file(filename, data):
    """Save YAML data to a file."""
    with open(filename, "w") as file:
        yaml.dump(data, file)

def create_backup(filename):
    """Create a timestamped backup of the file."""
    # create a directory called backups relative to the filename
    backup_dir = filename.parent / "yaml_backups"
    try:
        backup_dir.mkdir(exist_ok=True)
    except Exception as e:
        print(
            f"Error: Could not create backup directory {backup_dir}. Check your file-permissions or use --no-create-backup to skip creating a backup."
        )
        exit(2)

    timestamp = datetime.now().strftime("%y%m%d%H%M%S")
    backup_filename = backup_dir / f"{filename.name}.{timestamp}.bak"
    shutil.copyfile(filename, backup_filename)
    print(f"Backup created: {backup_filename}")

def main():
    parser = argparse.ArgumentParser(
        description="Deep merge YAML contents of src into dest."
    )
    parser.add_argument("src", type=Path, help="Source filename")
    parser.add_argument("dest", type=Path, help="Destination filename")
    parser.add_argument(
        "--create-backup",
        type=bool,
        default=True,
        help="Create a backup of the destination file before merging",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Print to stdout only, do not write to the destination file",
    )
    # add a argument for showing the stack trace on yaml parse errors
    parser.add_argument(
        "--show-stacktrace",
        action="store_true",
        help="Show stack trace on yaml parse errors",
    )
    # add an argument to clobber lists
    parser.add_argument(
        "--merge-lists",
        action="store_true",
        help="Merge list items instead of clobbering",
        default=False,
    )

    args = parser.parse_args()

    src_filename = args.src.resolve().expanduser()
    dest_filename = args.dest.resolve().expanduser()

    # make sure both files exist
    if not src_filename.exists():
        print(f"Error: {args.src} does not exist")
        exit(2)

    if not dest_filename.exists():
        print(f"Error: {args.dest} does not exist")
        exit(2)

    try:
        src_data = load_yaml_file(src_filename)
    except Exception as e:
        print(
            f"Error: {args.src} is not a valid YAML file. Run with --show-stacktrace to see the error."
        )
        if args.show_stacktrace:
            raise e
        exit(2)
    try:
        dest_data = load_yaml_file(dest_filename)
    except Exception as e:
        print(
            f"Error: {args.dest} is not a valid YAML file. Run with --show-stacktrace to see the error."
        )
        if args.show_stacktrace:
            raise e
        exit(2)

    if args.create_backup and not args.dry_run:
        create_backup(dest_filename)

    src_data = load_yaml_file(args.src)
    dest_data = load_yaml_file(args.dest)

    # if dest_data is empty, just copy src_data to dest_data
    if not dest_data:
        if not args.dry_run:
            save_yaml_file(args.dest, src_data)
    else:
        merged_data = deep_merge(dest_data, src_data, merge_lists=args.merge_lists)
        if not args.dry_run:
            save_yaml_file(args.dest, merged_data)
            print(f"Merged data from {args.src} into {args.dest}")
        else:
            yaml.dump(merged_data, sys.stdout)

if __name__ == "__main__":
    main()

```