/
project.py
279 lines (227 loc) · 10.2 KB
/
project.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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
# -*- coding: utf-8 -*-
"""Contain the logic for conda environment creation from ``meta.yaml``.
cenv is a tool to handle conda environment creation and update from the
dependency-definition inside the ``meta.yaml`` file.
As default conda has two files for dependency management:
* the ``environment.yml``
* and the ``meta.yaml``
In the ``environment.yml`` the environment-definition is stored.
In the ``meta.yaml`` the required information to build a conda-package are
stored.
This means redundant information.
cenv collects the dependency-information and all project-specific settings
from the ``meta.yaml``.
The collected information is used to create / update the projects conda
environment.
"""
from argparse import ArgumentParser
from argparse import Namespace
from argparse import RawTextHelpFormatter
from pathlib import Path
from typing import List
from typing import NoReturn
import attr
from cenv_tool import __version__
from cenv_tool.rules import ARGPARSE_DESCRIPTION
from cenv_tool.rules import CondaCmdFormats
from cenv_tool.rules import RULES
from cenv_tool.rules import Rules
from cenv_tool.utils import CenvProcessError
from cenv_tool.utils import extract_dependencies_from_meta_yaml
from cenv_tool.utils import message
from cenv_tool.utils import read_config
from cenv_tool.utils import read_meta_yaml
from cenv_tool.utils import run_in_bash
@attr.s(slots=True, auto_attribs=True)
class Project:
"""Contain a python-project using conda environments.
Containing methods to display information to current project and methods
to update the projects conda-environment from the settings defined in the
projects ``meta.yaml``.
"""
rules: Rules
conda_folder: Path = attr.ib(default=None)
env_folder: Path = attr.ib(default=None)
env_name: str = attr.ib(default=None)
dependencies: dict = attr.ib(default=None)
is_env: bool = attr.ib(default=None)
export_environment_yml: bool = attr.ib(None)
cmds: CondaCmdFormats = attr.ib(default=None)
cmd_kwargs: dict = attr.ib(default=None)
is_git: bool = attr.ib(default=None)
def __attrs_post_init__(self):
"""Set the more complex attributes of the project class."""
try:
meta_yaml = read_meta_yaml(Path.cwd())
except FileNotFoundError:
message(text='project has no meta.yaml!', color='red')
exit(1)
settings = meta_yaml['extra']['cenv']
dependencies = extract_dependencies_from_meta_yaml(meta_yaml)
config = read_config()
self.is_git = (Path.cwd() / self.rules.git_folder).exists()
self.export_environment_yml = config['export_environment_yml']
self.conda_folder = Path(config['conda_folder'])
self.env_folder = Path(config['env_folder'])
self.env_name = settings['env_name']
self.dependencies = dependencies
self.is_env = self.env_name in self.collect_available_envs()
conda_bin = self.rules.conda_cmds.conda_bin(self.conda_folder)
self.cmds = self.rules.conda_cmds
self.cmd_kwargs = {
'conda': conda_bin,
'name': self.env_name,
'pkgs': ' '.join([f'"{_}"' for _ in self.dependencies]),
}
def collect_available_envs(self) -> List[str]:
"""Collect the names of the conda environments currently installed.
Parameters:
conda_folder: the path where conda is installed.
Returns:
list of currently installed conda-environments
"""
return run_in_bash(
str(self.conda_folder.absolute()) +
'/bin/conda env list | awk \'{ if( !($1=="#") ) print $1 }\'',
).split('\n')
def write_new_md5sum(self):
"""Write new md5sum of ``meta.yaml`` to ``conda-build/meta.md5``."""
message(text='write md5sum of meta.yaml', color='bold', special='row')
command = (
'echo "$(md5sum $PWD/conda-build/meta.yaml)" | '
'cut -d\' \' -f1 > $PWD/conda-build/meta.md5'
)
run_in_bash(cmd=command)
message(text='updated', color='green', special='end', indent=2)
def export_environment_definition(self) -> NoReturn:
"""Export projects environment definition to an ``environment.yml``."""
message(text='Export environment.yml ...', color='bold', special='row')
run_in_bash(cmd=self.cmds.export.format(**self.cmd_kwargs))
message(text='Exported', color='green', special='end', indent=2)
def _remove_backup_environment(self) -> NoReturn:
"""Remove backup environment cloned from original environment."""
run_in_bash(cmd=self.cmds.clean.format(**self.cmd_kwargs))
def _restore_environment_from_backup(self, cloned: bool) -> NoReturn:
"""Restore the environment from the cloned backup environment.
After restore the backup environment is removed.
Parameters:
cloned: indicates if the environment already existed and a backup
was created.
"""
message(text='Error during creation!', color='red', special='row')
if self.is_env and cloned:
message(text='Recreating backup', color='bold', special='row')
run_in_bash(cmd=self.cmds.restore.format(**self.cmd_kwargs))
self._remove_backup_environment()
message(text='Recreated', color='green', special='end', indent=2)
message(text='Exit', color='red', special='end')
def _remove_previous_environment(self) -> NoReturn:
"""Remove old version of project environment.
If the old environment can't be removed, the backup made is removed.
"""
try:
message(text='Remove existing env', color='bold', special='row')
run_in_bash(cmd=self.cmds.remove.format(**self.cmd_kwargs))
message(text='Removed', color='green', special='end', indent=2)
except CenvProcessError:
self._remove_backup_environment()
message(
text=(
'Could not remove environment because it is '
'activated! Please deactivate it first.'
),
color='red',
)
exit(1)
def clone_environment_as_backup(self) -> NoReturn:
"""Clone the existing environment as a backup.
If the backup already exists, the previous backup is removed, then
the new one is created by cloning the current project environment.
"""
backup_name = f'{self.env_name}_backup'
if backup_name in self.collect_available_envs():
message(text='Clear old backup', color='bold', special='row')
self._remove_backup_environment()
message(text='Cleared', color='green', special='end', indent=2)
message(text='Create backup', color='bold', special='row')
run_in_bash(cmd=self.cmds.clone.format(**self.cmd_kwargs))
message(text='Created', color='green', special='end', indent=2)
def _handle_existing_environment(self) -> bool:
"""Check if environment already exists and create a backup of it."""
if self.is_env:
self.clone_environment_as_backup()
self._remove_previous_environment()
return True
return False
def create_environment(self, cloned: bool) -> NoReturn:
"""Create the environment for the project.
Try to create the environment for the project. If the environment
already existed and a backup was made and any error occure, restore the
backup environment.
If everything worked correctly finally remove the backup (if one was
made).
Parameters:
cloned: indicates if the environment already existed and a backup
was created.
"""
message(text='Create environment', color='bold', special='row')
try:
run_in_bash(cmd=self.cmds.create.format(**self.cmd_kwargs))
except CenvProcessError:
self._restore_environment_from_backup(cloned=cloned)
exit(1)
if cloned:
message(text='Clear backup', color='bold', special='row', indent=2)
run_in_bash(cmd=self.cmds.clean.format(**self.cmd_kwargs))
message(text='Cleared', color='green', special='end', indent=3)
message(text='Created', color='green', special='end', indent=2)
def update(self) -> NoReturn:
"""Create / recreate the conda environment of the current project.
If the conda environment already exists, clone the environment as a
backup and then remove original environment. Then create the new
conda environment. If a backup was created it is
removed afterwards. If any errors occurs during creation of the new
environment, recreate the old environment from backup and remove the
backup afterwards. If activated in the config-file, export the
environment-definition of the created environment to an
``environment.yml`` file. Finally store the md5sum of the meta.yaml for
the autoupdate feature.
"""
if self.is_env:
message(text=f'Updating {self.env_name}', color='cyan')
else:
message(text=f'Creating {self.env_name}', color='cyan')
cloned = self._handle_existing_environment()
self.create_environment(cloned=cloned)
if self.export_environment_yml:
self.export_environment_definition()
self.write_new_md5sum()
message(text='Done', color='green', special='end')
def _build_arguments() -> ArgumentParser:
"""Create arguments for the cenv-tool.
Returns:
the parsed arguments.
"""
parser = ArgumentParser(
description=ARGPARSE_DESCRIPTION,
epilog='For additional information see http://www.cenv.ouroboros.info',
formatter_class=RawTextHelpFormatter,
)
parser.add_argument(
'-v',
'--version',
action='store_true',
default=False,
help='Show current version of cenv and exit.',
)
return parser
def main() -> NoReturn:
"""Collect the required args, initialize and run the Project."""
parser = _build_arguments()
options = parser.parse_args()
if options.version:
print(__version__)
else:
Project(rules=RULES).update()
if __name__ == '__main__':
main()