-
Notifications
You must be signed in to change notification settings - Fork 1
/
requirements.py
59 lines (47 loc) · 1.91 KB
/
requirements.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
import hashlib
import shutil
import subprocess
from os import path
import logging
logger = logging.getLogger(__name__)
def get_md5_of_file(filepath):
try:
h = hashlib.md5()
with open(filepath, 'rb') as f:
h.update(f.read())
return h.hexdigest()
except IOError:
return None
class Requirements:
CACHE_PATH = "/packages"
def __init__(self, requirements_path, tag='default'):
self.tag = tag
self.requirements_path = requirements_path
self.cached_requirements_path = path.join(self.CACHE_PATH, f"{self.tag}_requirements.txt")
@property
def directory(self):
return self.get_requirements_directory(self.cached_requirements_path)
def get_requirements_directory(self, requirements_path):
md5_hash = get_md5_of_file(requirements_path)
if md5_hash:
return path.join(self.CACHE_PATH, md5_hash)
else:
return None
def ensure_installed(self, force_reinstall=False):
if not path.exists(self.requirements_path):
logger.info(f"No requirements.txt found, skipping package installation")
else:
if self.directory and path.exists(self.directory) and not force_reinstall:
logger.info(f"Requirements not changed, skipping update for tag '{self.tag}'...")
else:
self._install_packages()
def _install_packages(self):
logger.info(f"Updating requirements for tag '{self.tag}'...")
# If previous requirements exists, let's remove them
shutil.rmtree(self.directory, ignore_errors=True)
new_requirements_dir = self.get_requirements_directory(self.requirements_path)
subprocess.run([
"python", "-m", "pip", "install",
"-t", new_requirements_dir,
"-r", self.requirements_path])
shutil.copy(self.requirements_path, self.cached_requirements_path)