config_ninja.hooks

  1"""Integrate with `poethepoet` for callback hooks.
  2
  3## Example
  4
  5The following `config-ninja`_ settings file configures two local backends connected to three `poethepoet` hooks:
  6```yaml
  7.. include:: ../../examples/hooks.yaml
  8    :end-before: example-0
  9```
 10```yaml
 11  example-0:
 12.. include:: ../../examples/hooks.yaml
 13    :start-after: example-0:
 14    :end-before: example-1
 15```
 16```yaml
 17  example-1:
 18.. include:: ../../examples/hooks.yaml
 19    :start-after: example-1:
 20    :end-before: # define
 21```
 22```yaml
 23# define 'poethepoet' tasks in YAML (instead of TOML)
 24tool.poe:
 25  # ref https://poethepoet.natn.io/tasks/index.html
 26  tasks:
 27.. include:: ../../examples/hooks.yaml
 28    :start-after: tasks:
 29```
 30.. _config-ninja: https://config-ninja.readthedocs.io/home.html
 31"""
 32
 33from __future__ import annotations
 34
 35import logging
 36import os
 37import sys
 38from pathlib import Path
 39
 40# pyright: reportMissingTypeStubs=false
 41import poethepoet.context
 42import poethepoet.exceptions
 43from poethepoet import exceptions
 44from poethepoet.config import PoeConfig
 45from poethepoet.task.base import PoeTask, TaskContext, TaskSpecFactory
 46from poethepoet.task.graph import TaskExecutionGraph
 47from poethepoet.ui import PoeUi
 48
 49__all__ = ['Hook', 'HooksEngine', 'exceptions']
 50
 51logger = logging.getLogger(__name__)
 52
 53
 54class Hook:
 55    """Simple callable to execute the named hook with the given engine."""
 56
 57    engine: HooksEngine
 58    """The `HooksEngine` instance contains each of the hooks that can be executed."""
 59
 60    name: str
 61    """The name of the `poethepoet` task (`Hook`) to invoke."""
 62
 63    def __init__(self, engine: HooksEngine, name: str) -> None:
 64        """Raise a `ValueError` if the engine does not define a hook by the given name."""
 65        if name not in engine:
 66            raise ValueError(f'Undefined hook {name!r} (options: {list(engine.tasks)})')
 67
 68        self.name = name
 69        self.engine = engine
 70
 71    def __repr__(self) -> str:
 72        """The string representation of the `Hook` instance.
 73
 74        <!-- Perform doctest setup that is excluded from the docs
 75        >>> engine = ['example-hook']
 76
 77        -->
 78        >>> Hook(engine, 'example-hook')
 79        <Hook: 'example-hook'>
 80        """
 81        return f'<{self.__class__.__name__}: {self.name!r}>'
 82
 83    def __call__(self) -> None:
 84        """Invoke the `Hook` instance to execute it."""
 85        self.engine.execute(self.name)
 86
 87
 88class HooksEngine:
 89    """Encapsulate configuration for executing `poethepoet` tasks as callback hooks."""
 90
 91    config: PoeConfig
 92    """Contains `poethepoet` configuration settings."""
 93
 94    tasks: dict[str, PoeTask]
 95    """Name `poethepoet.task.base.PoeTask` objects for execution by name."""
 96
 97    ui: PoeUi
 98    """Used to parse configuration for running tasks."""
 99
100    def __init__(self, config: PoeConfig, ui: PoeUi, tasks: dict[str, PoeTask]) -> None:
101        """Initialize the engine with the given configuration, UI, and tasks."""
102        self.config = config
103        self.tasks = tasks
104        self.ui = ui
105
106    def __contains__(self, item: str) -> bool:
107        """Convenience method for checking if a hook is defined in the engine."""
108        return item in self.tasks
109
110    def get_hook(self, name: str) -> Hook:
111        """Initialize a `Hook` instance for running the `poethepoet.task.base.PoeTask` of the given name."""
112        return Hook(self, name)
113
114    def get_run_context(self, multistage: bool = False) -> poethepoet.context.RunContext:
115        """Create a `poethepoet.context.RunContext` instance for executing tasks.
116
117        This method is based on `poethepoet.app.PoeThePoet.get_run_context()` (`reference`_).
118
119        .. _reference: https://github.com/nat-n/poethepoet/blob/3c9fd8bcffde8a95c5cd9513923d0f43c1507385/poethepoet/app.py#L210-L225
120        """
121        return poethepoet.context.RunContext(
122            config=self.config,
123            ui=self.ui,
124            env=os.environ,
125            dry=self.ui['dry_run'] or False,
126            poe_active=os.environ.get('POE_ACTIVE'),
127            multistage=multistage,
128            cwd=Path.cwd(),
129        )
130
131    def execute(self, hook_name: str) -> None:
132        """Execute the `poethepoet.task.base.PoeTask` of the given name."""
133        self.ui.parse_args([hook_name])
134
135        task = self.tasks[hook_name]
136
137        if task.has_deps():
138            self.run_task_graph(task)
139        else:
140            self.run_task(task)
141
142    @classmethod
143    def load_file(cls, path: Path, default_paths: list[Path]) -> HooksEngine:
144        """Instantiate a `poethepoet.config.PoeConfig` object, then populate it with the given file."""
145        cfg = PoeConfig(config_name=tuple({str(p.name) for p in default_paths}))
146
147        cfg.load(path)
148        logger.debug('parsed hooks from %s: %s', path, list(cfg.task_names))
149
150        ui = PoeUi(
151            output=sys.stdout,
152            program_name=f'{sys.argv[0]} hook'
153            if sys.argv[0].endswith('config-ninja')
154            else f'{sys.executable} {sys.argv[0]} hook',
155        )
156
157        tasks: dict[str, PoeTask] = {}
158        factory = TaskSpecFactory(cfg)
159
160        for task_spec in factory.load_all():
161            task_spec.validate(cfg, factory)
162            tasks[task_spec.name] = task_spec.create_task(
163                invocation=(task_spec.name,),
164                ctx=TaskContext(config=cfg, cwd=str(task_spec.source.cwd), specs=factory, ui=ui),
165            )
166
167        return cls(cfg, ui, tasks)
168
169    def run_task(self, task: PoeTask, ctx: poethepoet.context.RunContext | None = None) -> None:
170        """Reimplement the `poethepoet.app.PoeThePoet.run_task()` method (`reference`_).
171
172        .. _reference: https://github.com/nat-n/poethepoet/blob/3c9fd8bcffde8a95c5cd9513923d0f43c1507385/poethepoet/app.py#L169-L181
173        """
174        try:
175            task.run(context=ctx or self.get_run_context())
176        except poethepoet.exceptions.ExecutionError as error:
177            logger.exception('error running task %s: %s', task.name, error)
178            raise
179
180    def run_task_graph(self, task: PoeTask) -> None:
181        """Reimplement the `poethepoet.app.PoeThePoet.run_task_graph()` method (`reference`_).
182
183        .. _reference: https://github.com/nat-n/poethepoet/blob/3c9fd8bcffde8a95c5cd9513923d0f43c1507385/poethepoet/app.py#L183-L208
184        """
185        ctx = self.get_run_context(multistage=True)
186        graph = TaskExecutionGraph(task, ctx)
187        plan = graph.get_execution_plan()
188
189        for stage in plan:
190            for stage_task in stage:
191                if stage_task == task:
192                    # The final sink task gets special treatment
193                    self.run_task(stage_task, ctx)
194                    return
195
196                task_result = stage_task.run(context=ctx)
197                if task_result:
198                    raise poethepoet.exceptions.ExecutionError(
199                        f'Task graph aborted after failed task {stage_task.name!r}'
200                    )
201
202
203logger.debug('successfully imported %s', __name__)