metadocker/src/metadocker/env/base.py

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)