-
-
Notifications
You must be signed in to change notification settings - Fork 52
/
context.py
95 lines (82 loc) · 2.71 KB
/
context.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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
import re
from pathlib import Path
from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, Tuple
from .env.manager import EnvVarsManager
from .executor import PoeExecutor
if TYPE_CHECKING:
from .config import PoeConfig
from .ui import PoeUi
class RunContext:
config: "PoeConfig"
ui: "PoeUi"
env: EnvVarsManager
dry: bool
poe_active: Optional[str]
project_dir: Path
multistage: bool = False
exec_cache: Dict[str, Any]
captured_stdout: Dict[Tuple[str, ...], str]
def __init__(
self,
config: "PoeConfig",
ui: "PoeUi",
env: Mapping[str, str],
dry: bool,
poe_active: Optional[str],
multistage: bool = False,
):
self.config = config
self.ui = ui
self.project_dir = Path(config.project_dir)
self.dry = dry
self.poe_active = poe_active
self.multistage = multistage
self.exec_cache = {}
self.captured_stdout = {}
self.env = EnvVarsManager(self.config, self.ui, base_env=env)
@property
def executor_type(self) -> Optional[str]:
return self.config.executor["type"]
def get_task_env(
self,
parent_env: Optional[EnvVarsManager],
task_envfile: Optional[str],
task_env: Optional[Mapping[str, str]],
task_uses: Optional[Mapping[str, Tuple[str, ...]]] = None,
) -> EnvVarsManager:
if parent_env is None:
parent_env = self.env
result = parent_env.for_task(task_envfile, task_env)
# Include env vars from dependencies
if task_uses is not None:
result.update(self.get_dep_values(task_uses))
return result
def get_dep_values(
self, used_task_invocations: Mapping[str, Tuple[str, ...]]
) -> Dict[str, str]:
"""
Get env vars from upstream tasks declared via the uses option.
New lines are replaced with whitespace similar to how unquoted command
interpolation works in bash.
"""
return {
var_name: re.sub(
r"\s+", " ", self.captured_stdout[invocation].strip("\r\n")
)
for var_name, invocation in used_task_invocations.items()
}
def get_executor(
self,
invocation: Tuple[str, ...],
env: EnvVarsManager,
task_options: Dict[str, Any],
) -> PoeExecutor:
return PoeExecutor.get(
invocation=invocation,
context=self,
env=env,
working_dir=self.project_dir / task_options.get("cwd", "."),
dry=self.dry,
executor_config=task_options.get("executor"),
capture_stdout=task_options.get("capture_stdout", False),
)