config_ninja.settings.poe

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