import json import sys import tempfile from pathlib import Path from metadocker.cmdline.base import ArgsFromFile from metadocker.common.errors import MetadockerError from metadocker.common.misc import get_root_dir from metadocker.db.db import settings from metadocker.docker import DockerBuilder from metadocker.common.var import VarDict, VarStr, VarPath, NotEmpty, Var, EmptyValue settings.load() class EnvHolder: def __init__(self): self._items = [] def __getitem__(self, item): return getattr(self, item) def __setitem__(self, key, value): if key not in self._items: self._items.append(key) return setattr(self, key, value) def items(self): return [ (k, getattr(self, k)) for k in self._items ] class Env: _current_env = None _suspend_run_ = False PROJECT : str PROJECT_VERSION : str DOCKER_TAG : str DOCKER_NAME : str ENV : dict PORTS : dict VOLUMES : dict COPY : dict APP_DIR : Path _vars_ = [ VarStr("PROJECT", required=True, help="Le nom du projet", is_meta=True, ask=True), VarStr("PROJECT_VERSION", None, help="La version du projet", validate=NotEmpty, is_meta=True, null=True, ask=True), VarStr("DOCKER_TAG", None, is_meta=True, validate=NotEmpty, help="Nom du tag de l'image docker. Par défaut: {env.PROJECT.lower()}-{PROJECT_VERSION}", ask=True), VarStr("DOCKER_NAME", None, is_meta=True, validate=NotEmpty, help="Nom du conteneur docker. Par défaut {env.PROJECT.lower()}", ask=True), VarDict("ENV", {}, is_meta=True, required=True, help="La liste des variable d'environnemnt à utiliser"), VarDict("PORTS", {8101: 8000}, required=True, is_meta=True, help="Le mapping des ports. En clé les ports de l'hote et en valeur les port du container"), VarDict("VOLUMES", {"./data": "/data"}, required=True, help="Le mapping des ports. En clé les ports de l'hote et en valeur les port du container", is_meta=True, null=True), VarDict("COPY", {}, required=True, is_meta=True, help="Dictionaire des fichier a copier: cle: fs hote valeur fs invite"), VarPath("APP_DIR", "/data", required=True, is_meta=True, validate=NotEmpty, help="Le dossier ou est lancé le projet", ask=True), ] _name_ = None _alias_ = [] class EnvException(MetadockerError): def __init__(self, errors): super().__init__(errors) self.errors = errors def get_error(self): return "\n".join(repr(x) for x in self.errors) def __init__(self, debug=False, **kwargs): Env._current_env = self self.root = get_root_dir() self.debug = debug self.meta = None self.docker : DockerBuilder self.env = None for k, v in kwargs.items(): setattr(self, k, v) def init(self): self.meta = self.get_meta() self.docker = DockerBuilder( self.meta.PROJECT, self.meta.PROJECT_VERSION, self.meta.DOCKER_TAG, self.meta.DOCKER_NAME, ) def read(self): self.init() self.env = self.get_env() def _validate_fields(self, filter=None): data= EnvHolder() errors = [] for var in self._vars_: if filter and not filter(var): continue try: value = var.run_validation(self) except Var.ByPassException: value = None except Var.VarException as err: errors.append(err) continue data[var.name] = value if errors: raise self.EnvException(errors) for k,v in data.items(): setattr(data, k, v) setattr(self, k, v) return data def get_meta(self): return self._validate_fields(lambda x: x.is_meta) def get_env(self): return self._validate_fields() def generate_docker_file(self ): raise NotImplementedError() def build_docker_file(self, output=None): if not output: temp = tempfile.TemporaryDirectory() output = Path(temp.name) / "Dockerfile" output = Path(output) output.parent.mkdir(exist_ok=True, parents=True) self.generate_docker_file() output.write_text(self.docker.get_file_content()) def create_empty_config(self, values=None, interractive=True): def _value(v, field): if v is EmptyValue: return f"# type in ({', '.join([x.__name__ for x in field.expected_types] or ['*'])})" if v is None: return "None" if isinstance(v, Path): v = str(v) if isinstance(v, (str, list, int, float, dict)): if v == {}: return "{\n}" if v == []: return "[\n]" return json.dumps(v, indent=2) return v def _help(v): return v.replace("\n", "\n#") values = values or {} if interractive: self.ask_for_data(values) data = [ "#!/bin/env python3", f"from metadocker.env import {self.__class__.__name__}", f"env = {self.__class__.__name__}()", f"" ] for var in self._vars_: if var.help: data.append(f"# {var.help}") value = _value(getattr(self, var.name, var.default_value), var) if var.required or var.ask: data.append(f"env.{var.name}={value}") else: data.append(f"# env.{var.name}={value}") data.append("") data.append("# lancement ") data.append("env.main()") return "\n".join(data) def main(self, args=None): if Env._suspend_run_: return self.init() ArgsFromFile.main(self, args) def ask_for_data(self, values = None): values = values or {} for var in self._vars_: if var.ask: kwargs = {} if var.name in values: kwargs["value"] = values[var.name] value = var.ask_value(**kwargs) setattr(self, var.name, value)