logging and refactoring

This commit is contained in:
root
2021-09-02 14:09:24 +02:00
parent e0b218f603
commit 63c8b94454
+120 -17
View File
@@ -1,48 +1,151 @@
#!/usr/bin/python3 #!/usr/bin/python3
from __future__ import annotations
from functools import lru_cache
from pathlib import Path from pathlib import Path
from subprocess import call from subprocess import call, check_output
from datetime import datetime from datetime import datetime
from typing import List
import logging
KEEP_DAYS = 3 KEEP_DAYS = 3
def main(): def main():
setup_logger()
build_images() build_images()
update_containers() update_containers()
prune() prune()
def build_images(): @lru_cache(maxsize=1)
images = read_file(Path('images.conf')) def get_today() -> str:
today = datetime.now().strftime('%Y%m%d') return datetime.now().strftime('%Y%m%d')
class Image:
def __init__(self, base_name: str):
self._base_name = base_name
@property
def qual_name(self) -> str:
return f'{self._base_name}-panthera'
@property
def latest(self) -> str:
return f'{self.qual_name}:latest'
@property
def today(self) -> str:
return f'{self.qual_name}:{get_today()}'
@classmethod
def from_file(cls, path: Path) -> list[Image]:
return [cls(name) for name in read_file(path)]
@property
def _work_dir(self) -> Path:
return Path(f'../images/{self._base_name}')
def build(self):
if self._todays_image_exists:
logging.info(f'don\'t need to build {self._base_name}')
else:
logging.info(f'building {self._base_name}')
self._build_image()
self._tag_image()
@property
def _todays_image_exists(self) -> bool:
for image in images_on_host():
if self.qual_name in image and get_today() in image:
return True
else:
return False
def _build_image(self):
call(['docker', 'build', '--no-cache', f'--tag={self.today}', '.'],
cwd=self._work_dir)
def _tag_image(self):
call(['docker', 'tag', f'{self.today}', f'{self.latest}'])
class Container:
def __init__(self, name: str):
self._name = name
@property
def name(self) -> str:
return self._name
@property
def _work_dir(self) -> Path:
return Path(f'../container/{self.name}')
def _pull_images(self) -> None:
logging.info(' pulling images')
call(['docker-compose', 'pull', '--ignore-pull-failures', '--quiet'],
cwd=self._work_dir)
def _up(self) -> None:
logging.info(' bringing up')
call(['docker-compose', 'up', '--detach'],
cwd=self._work_dir)
def update(self) -> None:
logging.info(f'container: {self.name}')
self._pull_images()
self._up()
@classmethod
def from_file(cls, path: Path) -> list[Container]:
return [cls(name) for name in read_file(path)]
def build_images() -> None:
logging.info('building images')
images = Image.from_file(Path('images.conf'))
for image in images: for image in images:
path = Path(f'../images/{image}') image.build()
call(['docker', 'build', '--no-cache', f'--tag={image}-panthera:{today}', '.'],
cwd=path)
def update_containers(): def update_containers() -> None:
containers = read_file(Path('container.conf')) logging.info('updating containers')
containers = Container.from_file(Path('container.conf'))
for container in containers: for container in containers:
path = Path(f'../container/{container}') container.update()
call(['docker-compose', 'pull'],
cwd=path)
call(['docker-compose', 'up', '-d'],
cwd=path)
def read_file(path: Path) -> List[str]: @lru_cache(maxsize=1)
def images_on_host() -> list[str]:
result = check_output(['docker', 'images']).decode('utf-8').strip()
images = result.split('\n')
for image in images:
logging.debug(image)
return images
def read_file(path: Path) -> list[str]:
content = path.read_text().strip() content = path.read_text().strip()
elements = content.split('\n') elements = content.split('\n')
elements = [item for item in elements if not item.startswith('#')] elements = [item for item in elements if not item.startswith('#')]
return elements return elements
def prune(): def prune() -> None:
logging.info('pruning')
hours = int(KEEP_DAYS * 24) hours = int(KEEP_DAYS * 24)
call(['docker', 'system', 'prune', '--all', '--force', '--filter', f'until={hours}h']) call(['docker', 'system', 'prune', '--all', '--force', '--filter', f'until={hours}h'])
def setup_logger() -> None:
logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.INFO)
logging.info('set up logger')
if __name__ == "__main__": if __name__ == "__main__":
main() main()