refactor: replace distutils' copy_tree with shutil copytree

.github/workflows/test.yml

@@ -18,7 +18,7 @@ jobs:
  runs-on: ubuntu-latest
  strategy:
  matrix:
- python-version: [3.7, 3.8, 3.9]
+ python-version: [3.8, 3.9]
 
  steps:
  - name: Checkout kapitan recursively

.pre-commit-config.yaml

@@ -3,7 +3,7 @@ repos:
  rev: 22.10.0
  hooks:
  - id: black
- language_version: python3.7
+ language_version: python3.8
  - repo: https:/igorshubovych/markdownlint-cli
  rev: v0.32.2
  hooks:

Dockerfile

@@ -1,5 +1,5 @@
 # Build the virtualenv for Kapitan
-FROM python:3.7-slim AS python-builder
+FROM python:3.8-slim AS python-builder
 
 ARG TARGETARCH
 
@@ -37,7 +37,7 @@ RUN curl -fsSL -o get_helm.sh https://raw.githubusercontent.com/helm/helm/master
  && rm get_helm.sh
 
 # Final image with virtualenv built in previous step
-FROM python:3.7-slim
+FROM python:3.8-slim
 
 COPY --from=python-builder /opt/venv /opt/venv
 

kapitan/initialiser.py

@@ -9,30 +9,34 @@
 
 import logging
 import os
-from distutils.dir_util import copy_tree
+from pathlib import Path
+
+from kapitan.utils import copy_tree
 
 logger = logging.getLogger(__name__)
 
 
 def initialise_skeleton(args):
- """Initialises a directory with a recommended skeleton structure
+ """Initialises a directory with a recommended skeleton structure and prints list of
+ copied files.
+
  Args:
- args.directory (string): path which to initialise, directory is assumed to exist
+ args.directory (string): path which to initialise, directory is assumed to exist.
  """
+ templates = Path(__file__).resolve().parent.joinpath("inputs", "templates")
+ destination = Path(args.directory)
+ copied_files = copy_tree(source=templates, destination=destination, dirs_exist_ok=True)
 
- current_pwd = os.path.dirname(__file__)
- templates_directory = os.path.join(current_pwd, "inputs", "templates")
- populated = copy_tree(templates_directory, args.directory)
  logger.info("Populated %s with:", args.directory)
- for directory, subs, file_list in os.walk(args.directory):
+ for directory, _, file_list in os.walk(args.directory):
  # In order to avoid adding the given directory itself in listing.
  if directory == args.directory:
  continue
- if any([path.startswith(directory) for path in populated]):
+ if any([path.startswith(directory) for path in copied_files]):
  level = directory.replace(args.directory, "").count(os.sep) - 1
  indent = " " * 4 * (level)
  logger.info("%s%s", indent, os.path.basename(directory))
  for fname in file_list:
- if os.path.join(directory, fname) in populated:
+ if os.path.join(directory, fname) in copied_files:
  sub_indent = " " * 4 * (level + 1)
  logger.info("%s%s", sub_indent, fname)

kapitan/targets.py

@@ -169,9 +169,10 @@ def compile_targets(
 
  # if '-t' is set on compile or only a few changed, only override selected targets
  if updated_targets:
- for target in updated_targets:
- compile_path_target = os.path.join(compile_path, target)
- temp_path_target = os.path.join(temp_compile_path, target)
+ for target in target_objs:
+ path = target["target_full_path"]
+ compile_path_target = os.path.join(compile_path, path)
+ temp_path_target = os.path.join(temp_compile_path, path)
 
  os.makedirs(compile_path_target, exist_ok=True)
 
@@ -404,7 +405,7 @@ def load_target_inventory(inventory_path, targets, ignore_class_notfound=False):
  raise InventoryError(
  "InventoryError: {}: parameters.kapitan has no assignment".format(target_name)
  )
- target_obj["target_full_path"] = inv_target["__reclass__"]["node"].replace("./", "")
+ target_obj["target_full_path"] = inv_target["parameters"]["_reclass_"]["name"]["path"]
  require_compile = not ignore_class_notfound
  valid_target_obj(target_obj, require_compile)
  validate_matching_target_name(target_name, target_obj, inventory_path)

kapitan/utils.py

@@ -23,6 +23,9 @@
 from distutils.file_util import _copy_file_contents
 from functools import lru_cache, wraps
 from hashlib import sha256
+from pathlib import Path
+from shutil import copytree
+from typing import List
 from zipfile import ZipFile
 
 import jinja2
@@ -631,3 +634,19 @@ def safe_copy_tree(src, dst):
  outputs.append(dst_name)
 
  return outputs
+
+
+def copy_tree(source: Path, destination: Path, dirs_exist_ok: bool = False) -> List[str]:
+ """Recursively copy a given directory from `source` to `destination` using shutil.copytree
+ and returns list of copied files. When `dirs_exist_ok` is set, the `FileExistsError` is
+ ignore when destination directory exists.
+ Args:
+ source (str): Path to a source directory
+ destination (str): Path to a destination directory
+ Returns:
+ list[str]: List of copied files
+ """
+ inventory_before = list(destination.rglob("*"))
+ copytree(source, destination, dirs_exist_ok=dirs_exist_ok)
+ inventory_after = list(destination.rglob("*"))
+ return [str(d) for d in inventory_after if d not in inventory_before]