| # -*- coding: UTF-8 -*- |
| """ |
| Provides cleanup tasks for invoke build scripts (as generic invoke tasklet). |
| Simplifies writing common, composable and extendable cleanup tasks. |
| |
| PYTHON PACKAGE REQUIREMENTS: |
| * path.py >= 8.2.1 (as path-object abstraction) |
| * pathlib (for ant-like wildcard patterns; since: python > 3.5) |
| * pycmd (required-by: clean_python()) |
| |
| clean task: Add Additional Directories and Files to be removed |
| ------------------------------------------------------------------------------- |
| |
| Create an invoke configuration file (YAML of JSON) with the additional |
| configuration data: |
| |
| .. code-block:: yaml |
| |
| # -- FILE: invoke.yaml |
| # USE: clean.directories, clean.files to override current configuration. |
| clean: |
| extra_directories: |
| - **/tmp/ |
| extra_files: |
| - **/*.log |
| - **/*.bak |
| |
| |
| Registration of Cleanup Tasks |
| ------------------------------ |
| |
| Other task modules often have an own cleanup task to recover the clean state. |
| The :meth:`clean` task, that is provided here, supports the registration |
| of additional cleanup tasks. Therefore, when the :meth:`clean` task is executed, |
| all registered cleanup tasks will be executed. |
| |
| EXAMPLE:: |
| |
| # -- FILE: tasks/docs.py |
| from __future__ import absolute_import |
| from invoke import task, Collection |
| from tasklet_cleanup import cleanup_tasks, cleanup_dirs |
| |
| @task |
| def clean(ctx, dry_run=False): |
| "Cleanup generated documentation artifacts." |
| cleanup_dirs(["build/docs"]) |
| |
| namespace = Collection(clean) |
| ... |
| |
| # -- REGISTER CLEANUP TASK: |
| cleanup_tasks.add_task(clean, "clean_docs") |
| cleanup_tasks.configure(namespace.configuration()) |
| """ |
| |
| from __future__ import absolute_import, print_function |
| import os.path |
| import sys |
| import pathlib |
| from invoke import task, Collection |
| from invoke.executor import Executor |
| from invoke.exceptions import Exit, Failure, UnexpectedExit |
| from path import Path |
| |
| |
| # ----------------------------------------------------------------------------- |
| # CLEANUP UTILITIES: |
| # ----------------------------------------------------------------------------- |
| def cleanup_accept_old_config(ctx): |
| ctx.cleanup.directories.extend(ctx.clean.directories or []) |
| ctx.cleanup.extra_directories.extend(ctx.clean.extra_directories or []) |
| ctx.cleanup.files.extend(ctx.clean.files or []) |
| ctx.cleanup.extra_files.extend(ctx.clean.extra_files or []) |
| |
| ctx.cleanup_all.directories.extend(ctx.clean_all.directories or []) |
| ctx.cleanup_all.extra_directories.extend(ctx.clean_all.extra_directories or []) |
| ctx.cleanup_all.files.extend(ctx.clean_all.files or []) |
| ctx.cleanup_all.extra_files.extend(ctx.clean_all.extra_files or []) |
| |
| |
| def execute_cleanup_tasks(ctx, cleanup_tasks, dry_run=False): |
| """Execute several cleanup tasks as part of the cleanup. |
| |
| REQUIRES: ``clean(ctx, dry_run=False)`` signature in cleanup tasks. |
| |
| :param ctx: Context object for the tasks. |
| :param cleanup_tasks: Collection of cleanup tasks (as Collection). |
| :param dry_run: Indicates dry-run mode (bool) |
| """ |
| # pylint: disable=redefined-outer-name |
| executor = Executor(cleanup_tasks, ctx.config) |
| failure_count = 0 |
| for cleanup_task in cleanup_tasks.tasks: |
| try: |
| print("CLEANUP TASK: %s" % cleanup_task) |
| executor.execute((cleanup_task, dict(dry_run=dry_run))) |
| except (Exit, Failure, UnexpectedExit) as e: |
| print("FAILURE in CLEANUP TASK: %s (GRACEFULLY-IGNORED)" % cleanup_task) |
| failure_count += 1 |
| |
| if failure_count: |
| print("CLEANUP TASKS: %d failure(s) occured" % failure_count) |
| |
| |
| def cleanup_dirs(patterns, dry_run=False, workdir="."): |
| """Remove directories (and their contents) recursively. |
| Skips removal if directories does not exist. |
| |
| :param patterns: Directory name patterns, like "**/tmp*" (as list). |
| :param dry_run: Dry-run mode indicator (as bool). |
| :param workdir: Current work directory (default=".") |
| """ |
| current_dir = Path(workdir) |
| python_basedir = Path(Path(sys.executable).dirname()).joinpath("..").abspath() |
| warn2_counter = 0 |
| for dir_pattern in patterns: |
| for directory in path_glob(dir_pattern, current_dir): |
| directory2 = directory.abspath() |
| if sys.executable.startswith(directory2): |
| # pylint: disable=line-too-long |
| print("SKIP-SUICIDE: '%s' contains current python executable" % directory) |
| continue |
| elif directory2.startswith(python_basedir): |
| # -- PROTECT CURRENTLY USED VIRTUAL ENVIRONMENT: |
| if warn2_counter <= 4: |
| print("SKIP-SUICIDE: '%s'" % directory) |
| warn2_counter += 1 |
| continue |
| |
| if not directory.isdir(): |
| print("RMTREE: %s (SKIPPED: Not a directory)" % directory) |
| continue |
| |
| if dry_run: |
| print("RMTREE: %s (dry-run)" % directory) |
| else: |
| print("RMTREE: %s" % directory) |
| directory.rmtree_p() |
| |
| |
| def cleanup_files(patterns, dry_run=False, workdir="."): |
| """Remove files or files selected by file patterns. |
| Skips removal if file does not exist. |
| |
| :param patterns: File patterns, like "**/*.pyc" (as list). |
| :param dry_run: Dry-run mode indicator (as bool). |
| :param workdir: Current work directory (default=".") |
| """ |
| current_dir = Path(workdir) |
| python_basedir = Path(Path(sys.executable).dirname()).joinpath("..").abspath() |
| error_message = None |
| error_count = 0 |
| for file_pattern in patterns: |
| for file_ in path_glob(file_pattern, current_dir): |
| if file_.abspath().startswith(python_basedir): |
| # -- PROTECT CURRENTLY USED VIRTUAL ENVIRONMENT: |
| continue |
| if not file_.isfile(): |
| print("REMOVE: %s (SKIPPED: Not a file)" % file_) |
| continue |
| |
| if dry_run: |
| print("REMOVE: %s (dry-run)" % file_) |
| else: |
| print("REMOVE: %s" % file_) |
| try: |
| file_.remove_p() |
| except os.error as e: |
| message = "%s: %s" % (e.__class__.__name__, e) |
| print(message + " basedir: "+ python_basedir) |
| error_count += 1 |
| if not error_message: |
| error_message = message |
| if False and error_message: |
| class CleanupError(RuntimeError): |
| pass |
| raise CleanupError(error_message) |
| |
| |
| def path_glob(pattern, current_dir=None): |
| """Use pathlib for ant-like patterns, like: "**/*.py" |
| |
| :param pattern: File/directory pattern to use (as string). |
| :param current_dir: Current working directory (as Path, pathlib.Path, str) |
| :return Resolved Path (as path.Path). |
| """ |
| if not current_dir: |
| current_dir = pathlib.Path.cwd() |
| elif not isinstance(current_dir, pathlib.Path): |
| # -- CASE: string, path.Path (string-like) |
| current_dir = pathlib.Path(str(current_dir)) |
| |
| for p in current_dir.glob(pattern): |
| yield Path(str(p)) |
| |
| |
| # ----------------------------------------------------------------------------- |
| # GENERIC CLEANUP TASKS: |
| # ----------------------------------------------------------------------------- |
| @task |
| def clean(ctx, dry_run=False): |
| """Cleanup temporary dirs/files to regain a clean state.""" |
| cleanup_accept_old_config(ctx) |
| directories = ctx.cleanup.directories or [] |
| directories.extend(ctx.cleanup.extra_directories or []) |
| files = ctx.cleanup.files or [] |
| files.extend(ctx.cleanup.extra_files or []) |
| |
| # -- PERFORM CLEANUP: |
| execute_cleanup_tasks(ctx, cleanup_tasks, dry_run=dry_run) |
| cleanup_dirs(directories, dry_run=dry_run) |
| cleanup_files(files, dry_run=dry_run) |
| |
| |
| @task(name="all", aliases=("distclean",)) |
| def clean_all(ctx, dry_run=False): |
| """Clean up everything, even the precious stuff. |
| NOTE: clean task is executed first. |
| """ |
| cleanup_accept_old_config(ctx) |
| directories = ctx.config.cleanup_all.directories or [] |
| directories.extend(ctx.config.cleanup_all.extra_directories or []) |
| files = ctx.config.cleanup_all.files or [] |
| files.extend(ctx.config.cleanup_all.extra_files or []) |
| |
| # -- PERFORM CLEANUP: |
| # HINT: Remove now directories, files first before cleanup-tasks. |
| cleanup_dirs(directories, dry_run=dry_run) |
| cleanup_files(files, dry_run=dry_run) |
| execute_cleanup_tasks(ctx, cleanup_all_tasks, dry_run=dry_run) |
| clean(ctx, dry_run=dry_run) |
| |
| |
| @task(name="python") |
| def clean_python(ctx, dry_run=False): |
| """Cleanup python related files/dirs: *.pyc, *.pyo, ...""" |
| # MAYBE NOT: "**/__pycache__" |
| cleanup_dirs(["build", "dist", "*.egg-info", "**/__pycache__"], |
| dry_run=dry_run) |
| if not dry_run: |
| ctx.run("py.cleanup") |
| cleanup_files(["**/*.pyc", "**/*.pyo", "**/*$py.class"], dry_run=dry_run) |
| |
| |
| # ----------------------------------------------------------------------------- |
| # TASK CONFIGURATION: |
| # ----------------------------------------------------------------------------- |
| CLEANUP_EMPTY_CONFIG = { |
| "directories": [], |
| "files": [], |
| "extra_directories": [], |
| "extra_files": [], |
| } |
| def make_cleanup_config(**kwargs): |
| config_data = CLEANUP_EMPTY_CONFIG.copy() |
| config_data.update(kwargs) |
| return config_data |
| |
| |
| namespace = Collection(clean_all, clean_python) |
| namespace.add_task(clean, default=True) |
| namespace.configure({ |
| "cleanup": make_cleanup_config( |
| files=["*.bak", "*.log", "*.tmp", "**/.DS_Store", "**/*.~*~"] |
| ), |
| "cleanup_all": make_cleanup_config( |
| directories=[".venv*", ".tox", "downloads", "tmp"] |
| ), |
| # -- BACKWARD-COMPATIBLE: OLD-STYLE |
| "clean": CLEANUP_EMPTY_CONFIG.copy(), |
| "clean_all": CLEANUP_EMPTY_CONFIG.copy(), |
| }) |
| |
| |
| # -- EXTENSION-POINT: CLEANUP TASKS (called by: clean, clean_all task) |
| # NOTE: Can be used by other tasklets to register cleanup tasks. |
| cleanup_tasks = Collection("cleanup_tasks") |
| cleanup_all_tasks = Collection("cleanup_all_tasks") |
| |
| # -- EXTEND NORMAL CLEANUP-TASKS: |
| # DISABLED: cleanup_tasks.add_task(clean_python) |
| # |
| # ----------------------------------------------------------------------------- |
| # EXTENSION-POINT: CONFIGURATION HELPERS: Can be used from other task modules |
| # ----------------------------------------------------------------------------- |
| def config_add_cleanup_dirs(directories): |
| # pylint: disable=protected-access |
| the_cleanup_directories = namespace._configuration["clean"]["directories"] |
| the_cleanup_directories.extend(directories) |
| |
| def config_add_cleanup_files(files): |
| # pylint: disable=protected-access |
| the_cleanup_files = namespace._configuration["clean"]["files"] |
| the_cleanup_files.extend(files) |