poethepoet.config

 23class PoeConfig:
 24    _project_config: ProjectConfig
 25    _included_config: List[IncludedConfig]
 26
 27    """
 28    The filenames to look for when loading config
 29    """
 30    _config_filenames: Tuple[str, ...] = (
 31        "pyproject.toml",
 32        "poe_tasks.toml",
 33        "poe_tasks.yaml",
 34        "poe_tasks.json",
 35    )
 36    """
 37    The parent directory of the project config file
 38    """
 39    _project_dir: Path
 40    """
 41    This can be overridden, for example to align with poetry
 42    """
 43    _baseline_verbosity: int = 0
 44
 45    def __init__(
 46        self,
 47        cwd: Optional[Union[Path, str]] = None,
 48        table: Optional[Mapping[str, Any]] = None,
 49        config_name: Optional[Union[str, Sequence[str]]] = None,
 50    ):
 51        if config_name is not None:
 52            if isinstance(config_name, str):
 53                self._config_filenames = (config_name,)
 54            else:
 55                self._config_filenames = tuple(config_name)
 56
 57        self._project_dir = Path().resolve() if cwd is None else Path(cwd)
 58        self._project_config = ProjectConfig(
 59            {"tool.poe": table or {}}, path=self._project_dir, strict=False
 60        )
 61        self._included_config = []
 62
 63    def lookup_task(
 64        self, name: str
 65    ) -> Union[Tuple[Mapping[str, Any], ConfigPartition], Tuple[None, None]]:
 66        task = self._project_config.get("tasks", {}).get(name, None)
 67        if task is not None:
 68            return task, self._project_config
 69
 70        for include in reversed(self._included_config):
 71            task = include.get("tasks", {}).get(name, None)
 72            if task is not None:
 73                return task, include
 74
 75        return None, None
 76
 77    def partitions(self, included_first=True) -> Iterator[ConfigPartition]:
 78        if not included_first:
 79            yield self._project_config
 80        yield from self._included_config
 81        if included_first:
 82            yield self._project_config
 83
 84    @property
 85    def executor(self) -> Mapping[str, Any]:
 86        return self._project_config.options.executor
 87
 88    @property
 89    def task_names(self) -> Iterator[str]:
 90        result = list(self._project_config.get("tasks", {}).keys())
 91        for config_part in self._included_config:
 92            for task_name in config_part.get("tasks", {}).keys():
 93                # Don't use a set to dedup because we want to preserve task order
 94                if task_name not in result:
 95                    result.append(task_name)
 96        yield from result
 97
 98    @property
 99    def tasks(self) -> Dict[str, Any]:
100        result = dict(self._project_config.get("tasks", {}))
101        for config in self._included_config:
102            for task_name, task_def in config.get("tasks", {}).items():
103                if task_name in result:
104                    continue
105                result[task_name] = task_def
106        return result
107
108    @property
109    def default_task_type(self) -> str:
110        return self._project_config.options.default_task_type
111
112    @property
113    def default_array_task_type(self) -> str:
114        return self._project_config.options.default_array_task_type
115
116    @property
117    def default_array_item_task_type(self) -> str:
118        return self._project_config.options.default_array_item_task_type
119
120    @property
121    def shell_interpreter(self) -> Tuple[str, ...]:
122        raw_value = self._project_config.options.shell_interpreter
123        if isinstance(raw_value, list):
124            return tuple(raw_value)
125        return (raw_value,)
126
127    @property
128    def verbosity(self) -> int:
129        return self._project_config.get("verbosity", self._baseline_verbosity)
130
131    @property
132    def is_poetry_project(self) -> bool:
133        return (
134            self._project_config.path.name == "pyproject.toml"
135            and "poetry" in self._project_config.full_config.get("tool", {})
136        )
137
138    @property
139    def project_dir(self) -> Path:
140        return self._project_dir
141
142    def load(self, target_path: Optional[Union[Path, str]] = None, strict: bool = True):
143        """
144        target_path is the path to a file or directory for loading config
145        If strict is false then some errors in the config structure are tolerated
146        """
147
148        for config_file in PoeConfigFile.find_config_files(
149            target_path=Path(target_path or self._project_dir),
150            filenames=self._config_filenames,
151            search_parent=not target_path,
152        ):
153            config_file.load()
154
155            if config_file.error:
156                raise config_file.error
157
158            elif config_file.is_valid:
159                self._project_dir = config_file.path.parent
160
161                config_content = config_file.load()
162                assert config_content
163
164                try:
165                    self._project_config = ProjectConfig(
166                        config_content,
167                        path=config_file.path,
168                        project_dir=self._project_dir,
169                        strict=strict,
170                    )
171                except ConfigValidationError:
172                    # Try again to load Config with minimal validation so we can still
173                    # display the task list alongside the error
174                    self._project_config = ProjectConfig(
175                        config_content,
176                        path=config_file.path,
177                        project_dir=self._project_dir,
178                        strict=False,
179                    )
180                    raise
181
182                break
183
184        else:
185            raise PoeException(
186                f"No poe configuration found from location {target_path}"
187            )
188
189        self._load_includes(strict=strict)
190
191    def _load_includes(self: "PoeConfig", strict: bool = True):
192        # Attempt to load each of the included configs
193        for include in self._project_config.options.include:
194            include_path = self._resolve_include_path(include["path"])
195
196            if not include_path.exists():
197                # TODO: print warning in verbose mode, requires access to ui somehow
198                #       Maybe there should be something like a WarningService?
199
200                if POE_DEBUG:
201                    print(f" ! Could not include file from invalid path {include_path}")
202                continue
203
204            try:
205                config_file = PoeConfigFile(include_path)
206                config_content = config_file.load()
207                assert config_content
208
209                self._included_config.append(
210                    IncludedConfig(
211                        config_content,
212                        path=config_file.path,
213                        project_dir=self._project_dir,
214                        cwd=(
215                            self.project_dir.joinpath(include["cwd"]).resolve()
216                            if include.get("cwd")
217                            else None
218                        ),
219                        strict=strict,
220                    )
221                )
222                if POE_DEBUG:
223                    print(f"  Included config from {include_path}")
224            except (PoeException, KeyError) as error:
225                raise ConfigValidationError(
226                    f"Invalid content in included file from {include_path}",
227                    filename=str(include_path),
228                ) from error
229
230    def _resolve_include_path(self, include_path: str):
231        from ..env.template import apply_envvars_to_template
232
233        available_vars = {"POE_ROOT": str(self._project_dir)}
234
235        if "${POE_GIT_DIR}" in include_path:
236            from ..helpers.git import GitRepo
237
238            git_repo = GitRepo(self._project_dir)
239            available_vars["POE_GIT_DIR"] = str(git_repo.path or "")
240
241        if "${POE_GIT_ROOT}" in include_path:
242            from ..helpers.git import GitRepo
243
244            git_repo = GitRepo(self._project_dir)
245            available_vars["POE_GIT_ROOT"] = str(git_repo.main_path or "")
246
247        include_path = apply_envvars_to_template(
248            include_path, available_vars, require_braces=True
249        )
250
251        return self._project_dir.joinpath(include_path).resolve()

28class ConfigPartition:
29    options: PoeOptions
30    full_config: Mapping[str, Any]
31    poe_options: Mapping[str, Any]
32    path: Path
33    project_dir: Path
34    _cwd: Optional[Path]
35
36    ConfigOptions: Type[PoeOptions]
37    is_primary: bool = False
38
39    def __init__(
40        self,
41        full_config: Mapping[str, Any],
42        path: Path,
43        project_dir: Optional[Path] = None,
44        cwd: Optional[Path] = None,
45        strict: bool = True,
46    ):
47        self.poe_options: Mapping[str, Any] = (
48            full_config["tool"].get("poe", {})
49            if "tool" in full_config
50            else full_config.get("tool.poe", {})
51        )
52        self.options = next(
53            self.ConfigOptions.parse(
54                self.poe_options,
55                strict=strict,
56                # Allow and standard config keys, even if not declared
57                # This avoids misguided validation errors on included config
58                extra_keys=tuple(ProjectConfig.ConfigOptions.get_fields()),
59            )
60        )
61        self.full_config = full_config
62        self.path = path
63        self._cwd = cwd
64        self.project_dir = project_dir or self.path.parent
65
66    @property
67    def cwd(self):
68        return self._cwd or self.project_dir
69
70    @property
71    def config_dir(self):
72        return self._cwd or self.path.parent
73
74    def get(self, key: str, default: Any = NoValue):
75        return self.options.get(key, default)