poethepoet.app

  1import os
  2import sys
  3from pathlib import Path
  4from typing import (
  5    IO,
  6    TYPE_CHECKING,
  7    Any,
  8    Dict,
  9    Mapping,
 10    Optional,
 11    Sequence,
 12    Tuple,
 13    Union,
 14)
 15
 16from .exceptions import ExecutionError, PoeException
 17
 18if TYPE_CHECKING:
 19    from .config import PoeConfig
 20    from .context import RunContext
 21    from .task.base import PoeTask
 22    from .ui import PoeUi
 23
 24
 25class PoeThePoet:
 26    """
 27    :param cwd:
 28        The directory that poe should take as the current working directory,
 29        this determines where to look for a pyproject.toml file, defaults to
 30        ``Path().resolve()``
 31    :type cwd: Path, optional
 32    :param config:
 33        Either a dictionary with the same schema as a pyproject.toml file, or a
 34        `PoeConfig <https://github.com/nat-n/poethepoet/blob/main/poethepoet/config/config.py>`_
 35        object to use as an alternative to loading config from a file.
 36    :type config: dict | PoeConfig, optional
 37    :param output:
 38        A stream for the application to write its own output to, defaults to sys.stdout
 39    :type output: IO, optional
 40    :param poetry_env_path:
 41        The path to the poetry virtualenv. If provided then it is used by the
 42        `PoetryExecutor <https://github.com/nat-n/poethepoet/blob/main/poethepoet/executor/poetry.py>`_,
 43        instead of having to execute poetry in a subprocess to determine this.
 44    :type poetry_env_path: str, optional
 45    :param config_name:
 46        The name of the file to load tasks and configuration from. If not set then poe
 47        will search for config by the following file names: pyproject.toml
 48        poe_tasks.toml poe_tasks.yaml poe_tasks.json
 49    :type config_name: str, optional
 50    :param program_name:
 51        The name of the program that is being run. This is used primarily when
 52        outputting help messages, defaults to "poe"
 53    :type program_name: str, optional
 54    """
 55
 56    cwd: Path
 57    ui: "PoeUi"
 58    config: "PoeConfig"
 59
 60    _task_specs: Optional[Dict[str, "PoeTask.TaskSpec"]] = None
 61
 62    def __init__(
 63        self,
 64        cwd: Optional[Union[Path, str]] = None,
 65        config: Optional[Union[Mapping[str, Any], "PoeConfig"]] = None,
 66        output: IO = sys.stdout,
 67        poetry_env_path: Optional[str] = None,
 68        config_name: Optional[str] = None,
 69        program_name: str = "poe",
 70    ):
 71        from .config import PoeConfig
 72        from .ui import PoeUi
 73
 74        self.cwd = Path(cwd) if cwd else Path().resolve()
 75
 76        if self.cwd and self.cwd.is_file():
 77            config_name = self.cwd.name
 78            self.cwd = self.cwd.parent
 79
 80        self.config = (
 81            config
 82            if isinstance(config, PoeConfig)
 83            else PoeConfig(cwd=self.cwd, table=config, config_name=config_name)
 84        )
 85        self.ui = PoeUi(output=output, program_name=program_name)
 86        self._poetry_env_path = poetry_env_path
 87
 88    def __call__(self, cli_args: Sequence[str], internal: bool = False) -> int:
 89        """
 90        :param cli_args:
 91            A sequence of command line arguments to pass to poe (i.e. sys.argv[1:])
 92        :param internal:
 93            Indicates that this is an internal call to run poe, e.g. from a
 94            plugin hook.
 95        """
 96
 97        self.ui.parse_args(cli_args)
 98
 99        if self.ui["version"]:
100            self.ui.print_version()
101            return 0
102
103        try:
104            self.config.load(target_path=self.ui["project_root"])
105            for task_spec in self.task_specs.load_all():
106                task_spec.validate(self.config, self.task_specs)
107        except PoeException as error:
108            if self.ui["help"]:
109                self.print_help()
110                return 0
111            self.print_help(error=error)
112            return 1
113
114        self.ui.set_default_verbosity(self.config.verbosity)
115
116        if self.ui["help"]:
117            self.print_help()
118            return 0
119
120        task = self.resolve_task(internal)
121        if not task:
122            return 1
123
124        if task.has_deps():
125            return self.run_task_graph(task) or 0
126        else:
127            return self.run_task(task) or 0
128
129    @property
130    def task_specs(self):
131        if not self._task_specs:
132            from .task.base import TaskSpecFactory
133
134            self._task_specs = TaskSpecFactory(self.config)
135        return self._task_specs
136
137    def resolve_task(self, allow_hidden: bool = False) -> Optional["PoeTask"]:
138        from .task.base import TaskContext
139
140        task = tuple(self.ui["task"])
141        if not task:
142            self.print_help(info="No task specified.")
143            return None
144
145        task_name = task[0]
146        if task_name not in self.config.task_names:
147            self.print_help(error=PoeException(f"Unrecognised task {task_name!r}"))
148            return None
149
150        if task_name.startswith("_") and not allow_hidden:
151            self.print_help(
152                error=PoeException(
153                    "Tasks prefixed with `_` cannot be executed directly"
154                ),
155            )
156            return None
157
158        task_spec = self.task_specs.get(task_name)
159        return task_spec.create_task(
160            invocation=task,
161            ctx=TaskContext(
162                config=self.config,
163                cwd=str(task_spec.source.cwd),
164                specs=self.task_specs,
165                ui=self.ui,
166            ),
167        )
168
169    def run_task(
170        self, task: "PoeTask", context: Optional["RunContext"] = None
171    ) -> Optional[int]:
172        if context is None:
173            context = self.get_run_context()
174        try:
175            return task.run(context=context)
176        except ExecutionError as error:
177            self.ui.print_error(error=error)
178            return 1
179        except PoeException as error:
180            self.print_help(error=error)
181            return 1
182
183    def run_task_graph(self, task: "PoeTask") -> Optional[int]:
184        from .task.graph import TaskExecutionGraph
185
186        context = self.get_run_context(multistage=True)
187        graph = TaskExecutionGraph(task, context)
188        plan = graph.get_execution_plan()
189
190        for stage in plan:
191            for stage_task in stage:
192                if stage_task == task:
193                    # The final sink task gets special treatment
194                    return self.run_task(stage_task, context)
195
196                try:
197                    task_result = stage_task.run(context=context)
198                    if task_result:
199                        raise ExecutionError(
200                            f"Task graph aborted after failed task {stage_task.name!r}"
201                        )
202                except PoeException as error:
203                    self.print_help(error=error)
204                    return 1
205                except ExecutionError as error:
206                    self.ui.print_error(error=error)
207                    return 1
208        return 0
209
210    def get_run_context(self, multistage: bool = False) -> "RunContext":
211        from .context import RunContext
212
213        result = RunContext(
214            config=self.config,
215            ui=self.ui,
216            env=os.environ,
217            dry=self.ui["dry_run"],
218            poe_active=os.environ.get("POE_ACTIVE"),
219            multistage=multistage,
220            cwd=self.cwd,
221        )
222        if self._poetry_env_path:
223            # This allows the PoetryExecutor to use the venv from poetry directly
224            result.exec_cache["poetry_virtualenv"] = self._poetry_env_path
225        return result
226
227    def print_help(
228        self,
229        info: Optional[str] = None,
230        error: Optional[Union[str, PoeException]] = None,
231    ):
232        from .task.args import PoeTaskArgs
233
234        if isinstance(error, str):
235            error = PoeException(error)
236
237        tasks_help: Dict[
238            str, Tuple[str, Sequence[Tuple[Tuple[str, ...], str, str]]]
239        ] = {
240            task_name: (
241                (
242                    content.get("help", ""),
243                    PoeTaskArgs.get_help_content(content.get("args")),
244                )
245                if isinstance(content, dict)
246                else ("", tuple())
247            )
248            for task_name, content in self.config.tasks.items()
249        }
250
251        self.ui.print_help(tasks=tasks_help, info=info, error=error)