-
Notifications
You must be signed in to change notification settings - Fork 1.5k
/
clean.py
56 lines (46 loc) · 2.11 KB
/
clean.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
from pathlib import Path
from shutil import rmtree
from dbt import deprecations
from dbt.cli.flags import Flags
from dbt.config.project import Project
from dbt.events.types import CheckCleanPath, ConfirmCleanPath, FinishedCleanPaths
from dbt.task.base import BaseTask, move_to_nearest_project_dir
from dbt_common.events.functions import fire_event
from dbt_common.exceptions import DbtRuntimeError
class CleanTask(BaseTask):
def __init__(self, args: Flags, config: Project):
super().__init__(args)
self.config = config
self.project = config
def run(self):
"""
This function takes all the paths in the target file
and cleans the project paths that are not protected.
"""
project_dir = move_to_nearest_project_dir(self.args.project_dir)
potential_clean_paths = set(Path(p).resolve() for p in self.project.clean_targets)
source_paths = set(
Path(p).resolve() for p in (*self.project.all_source_paths, *self.project.test_paths)
)
clean_paths = potential_clean_paths.difference(source_paths)
if potential_clean_paths != clean_paths:
raise DbtRuntimeError(
f"dbt will not clean the following source paths: {[str(s) for s in source_paths.intersection(potential_clean_paths)]}"
)
paths_outside_project = set(
path for path in clean_paths if project_dir not in path.absolute().parents
)
if paths_outside_project and self.args.clean_project_files_only:
raise DbtRuntimeError(
f"dbt will not clean the following directories outside the project: {[str(p) for p in paths_outside_project]}"
)
if (
"dbt_modules" in self.project.clean_targets
and self.config.packages_install_path not in self.config.clean_targets
):
deprecations.warn("install-packages-path")
for path in clean_paths:
fire_event(CheckCleanPath(path=str(path)))
rmtree(path, True)
fire_event(ConfirmCleanPath(path=str(path)))
fire_event(FinishedCleanPaths())