/
poetry.py
133 lines (105 loc) · 4.31 KB
/
poetry.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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
from os import environ
from pathlib import Path
from typing import TYPE_CHECKING, Dict, Optional, Sequence, Type
from ..exceptions import ExecutionError
from .base import PoeExecutor
if TYPE_CHECKING:
from ..context import RunContext
class PoetryExecutor(PoeExecutor):
"""
A poe task executor implementation that executes inside a poetry managed dev
environment
"""
__key__ = "poetry"
__options__: Dict[str, Type] = {}
@classmethod
def works_with_context(cls, context: "RunContext") -> bool:
if not context.config.is_poetry_project:
return False
return bool(cls._poetry_cmd_from_path())
def execute(
self, cmd: Sequence[str], input: Optional[bytes] = None, use_exec: bool = False
) -> int:
"""
Execute the given cmd as a subprocess inside the poetry managed dev environment
"""
poetry_env = self._get_poetry_virtualenv()
if poetry_env:
from ..virtualenv import Virtualenv
# Execute the task in the virtualenv from poetry
venv = Virtualenv(Path(poetry_env))
return self._execute_cmd(
(venv.resolve_executable(cmd[0]), *cmd[1:]),
input=input,
env=venv.get_env_vars(self.env.to_dict()),
use_exec=use_exec,
)
if self._virtualenv_creation_disabled():
# There's no poetry env, and there isn't going to be
return self._execute_cmd(cmd, input=input, use_exec=use_exec)
# Run this task with `poetry run`
return self._execute_cmd(
(self._poetry_cmd(), "run", *cmd),
input=input,
use_exec=use_exec,
)
def _handle_file_not_found(
self, cmd: Sequence[str], error: FileNotFoundError
) -> int:
poetry_env = self._get_poetry_virtualenv()
error_context = f" using virtualenv {poetry_env!r}" if poetry_env else ""
raise ExecutionError(
f"executable {cmd[0]!r} could not be found{error_context}"
) from error
def _get_poetry_virtualenv(self, force: bool = True):
"""
Ask poetry where it put the virtualenv for this project.
Invoking poetry is relatively expensive so cache the result
"""
# TODO: see if there's a more efficient way to do this that doesn't involve
# invoking the poetry cli or relying on undocumented APIs
exec_cache = self.context.exec_cache
if force and "poetry_virtualenv" not in exec_cache:
from subprocess import PIPE, Popen
# Need to make sure poetry isn't influenced by whatever virtualenv is
# currently active
clean_env = dict(environ)
clean_env.pop("VIRTUAL_ENV", None)
clean_env["PYTHONIOENCODING"] = "utf-8"
exec_cache["poetry_virtualenv"] = (
Popen(
(self._poetry_cmd(), "env", "info", "-p"),
stdout=PIPE,
cwd=self.context.config.project_dir,
env=clean_env,
)
.communicate()[0]
.decode()
.strip()
)
return exec_cache.get("poetry_virtualenv")
@classmethod
def _poetry_cmd(cls):
from_path = cls._poetry_cmd_from_path()
if from_path:
return str(Path(from_path).resolve())
return "poetry"
@classmethod
def _poetry_cmd_from_path(cls):
import shutil
return shutil.which("poetry")
def _virtualenv_creation_disabled(self):
exec_cache = self.context.exec_cache
while "poetry_virtualenvs_create_disabled" not in exec_cache:
# Check env override
env_override = environ.get("POETRY_VIRTUALENVS_CREATE")
if env_override is not None:
exec_cache["poetry_virtualenvs_create_disabled"] = (
env_override == "false"
)
break
# A complete implementation would also check for a local poetry config file
# and a global poetry config file (location for this is platform dependent)
# in that order but just checking the env will do for now.
break
return exec_cache.get("poetry_virtualenvs_create_disabled", False)