tools.py

def create_path_filter(patterns: StrOrPathSeq) -> CheckPathFunc:
    """Returns a function that matches a path against given patterns."""
    patterns = [normalize_str(p) for p in patterns]
    spec = pathspec.PathSpec.from_lines("gitwildmatch", patterns)

    def match(path: StrOrPath) -> bool:
        return spec.match_file(str(path))

    return match

def get_jinja_env(
    envops: EnvOps,
    filters: Optional[Filters] = None,
    paths: Optional[LoaderPaths] = None,
    **kwargs: Any,
) -> SandboxedEnvironment:
    """Return a pre-configured Jinja environment."""
    loader = FileSystemLoader(paths) if paths else None
    # We want to minimize the risk of hidden malware in the templates
    # so we use the SandboxedEnvironment instead of the regular one.
    # Of couse we still have the post-copy tasks to worry about, but at least
    # they are more visible to the final user.
    env = SandboxedEnvironment(loader=loader, **envops.dict(), **kwargs)
    default_filters = {"to_nice_yaml": to_nice_yaml}
    default_filters.update(filters or {})
    env.filters.update(default_filters)
    return env

def get_migration_tasks(conf: ConfigData, stage: str) -> List[Dict]:
    """Get migration objects that match current version spec.

    Versions are compared using PEP 440.
    """
    result: List[Dict] = []
    if not conf.old_commit or not conf.commit:
        return result
    vfrom = version.parse(conf.old_commit)
    vto = version.parse(conf.commit)
    extra_env = {
        "STAGE": stage,
        "VERSION_FROM": conf.old_commit,
        "VERSION_TO": conf.commit,
    }
    for migration in conf.migrations:
        if vto >= version.parse(migration.version) > vfrom:
            extra_env = dict(extra_env, VERSION_CURRENT=str(migration.version))
            result += [
                {"task": task, "extra_env": extra_env}
                for task in migration.dict().get(stage, [])
            ]
    return result

def normalize_str(text: StrOrPath, form: str = "NFD") -> str:
    """Normalize unicode text. Uses the NFD algorithm by default."""
    return unicodedata.normalize(form, str(text))

def to_nice_yaml(data: Any, **kwargs) -> str:
    """Dump a string to pretty YAML."""
    # Remove security-problematic kwargs
    kwargs.pop("stream", None)
    kwargs.pop("Dumper", None)
    result = safe_dump(data, **kwargs)
    if isinstance(result, str):
        result = result.rstrip()
    return result or ""