majuna/app/terraform/list/__init__.py

from collections.abc import Mapping, Sequence
import json
import os
from typing import List, Any

from app import app
from app.lists import lists
from app.models.base import MirrorList, Pool
from app.terraform.terraform import TerraformAutomation


def obfuscator(obj: Any) -> Any:
    if isinstance(obj, str):
        return "".join([f"!AAA!{hex(ord(c))[2:].zfill(4)}" for c in obj])
    if isinstance(obj, Mapping):
        return {obfuscator(k): obfuscator(v) for k, v in obj.items()}
    if isinstance(obj, Sequence):
        return [obfuscator(i) for i in obj]
    return obj


def json_encode(obj: Any, obfuscate: bool) -> str:
    if obfuscate:
        obj = obfuscator(obj)
        result = json.dumps(obj, sort_keys=True).replace("!AAA!", "\\u")
        return result
    return json.dumps(obj, indent=2, sort_keys=True)


def javascript_encode(obj: Any, obfuscate: bool) -> str:
    return "var mirrors = " + json_encode(obj, obfuscate) + ";"


class ListAutomation(TerraformAutomation):
    template: str
    """
    Terraform configuration template using Jinja 2.
    """

    template_parameters: List[str]
    """
    List of parameters to be read from the application configuration for use
    in the templating of the Terraform configuration.
    """

    def tf_generate(self) -> None:
        if not self.working_dir:
            raise RuntimeError("No working directory specified.")
        self.tf_write(
            self.template,
            lists=MirrorList.query.filter(
                MirrorList.destroyed.is_(None),
                MirrorList.provider == self.provider,
            ).all(),
            global_namespace=app.config['GLOBAL_NAMESPACE'],
            terraform_modules_path=os.path.join(*list(os.path.split(app.root_path))[:-1], 'terraform-modules'),
            backend_config=f"""backend "http" {{
              lock_address = "{app.config['TFSTATE_BACKEND']}/{self.short_name}"
              unlock_address = "{app.config['TFSTATE_BACKEND']}/{self.short_name}"
              address = "{app.config['TFSTATE_BACKEND']}/{self.short_name}"
            }}""",
            **{
                k: app.config[k.upper()]
                for k in self.template_parameters
            }
        )
        for pool in Pool.query.filter(Pool.destroyed.is_(None)).all():
            for key, formatter in lists.items():
                for obfuscate in [True, False]:
                    with open(os.path.join(self.working_dir, f"{key}.{pool.pool_name}{'.jsno' if obfuscate else '.json'}"),
                              'w', encoding="utf-8") as out:
                        out.write(json_encode(formatter(pool), obfuscate))
                    with open(os.path.join(self.working_dir, f"{key}.{pool.pool_name}{'.jso' if obfuscate else '.js'}"),
                              'w', encoding="utf-8") as out:
                        out.write(javascript_encode(formatter(pool), obfuscate))