Stateless terraform apply

stateless_tf_apply.py

#! python
"""
This script is used to apply terraform changes to a given directory.

ANY STATE CHANGES STORED LOCALLY WILL BE LOST.

It is HIGHLY RECOMMENDED that you use this with a remote terraform backend.

Maintainer: Joshua Schlichting <[email protected]>
"""
import argparse
from asyncio.log import logger
from enum import Enum
import logging
import os
import shutil
from subprocess import Popen, PIPE
import sys
import tempfile
from typing import List


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class ProjectEnv(Enum):
    DEV = "dev"
    UAT = "uat"
    PRD = "prd"


def setup_temp_dir(*, template_tf_dir: str) -> tempfile.TemporaryDirectory:

    tmpdir = tempfile.TemporaryDirectory()
    tmpdirname = tmpdir.name
    print("created temporary directory", tmpdirname)
    print(os.getcwd())
    for file in os.listdir(template_tf_dir):
        if file.endswith(".tf"):
            shutil.copy(os.path.join("terraform", file), os.path.join(tmpdirname, file))
    return tmpdir


def replace_var(*, filename: str, var: str, value: str) -> None:
    with open(filename, "r") as file:
        original_main_tf_contents: List[str] = file.readlines()

    replaced_main_tf_contents: List[str] = []
    for line in original_main_tf_contents:
        formatted_var = "${var." + var + "}"
        if formatted_var in line:
            line = line.replace(formatted_var, value)
        replaced_main_tf_contents.append(line)
    with open(filename, "w") as file:
        file.writelines(replaced_main_tf_contents)
    return None


def is_tf_file(file: str) -> bool:
    return "tf" in file.split(".")[-1]


def run_tf_routine() -> None:
    commands: list = [
        ["terraform", "fmt", "-check"],
        ["terraform", "init"],
        ["terraform", "validate", "-no-color"],
        ["terraform", "plan", "-no-color"],
        ["terraform", "apply", "-auto-approve"],
    ]
    for command in commands:
        logger.info("Executing " + " ".join(command))
        process = Popen(command, stdout=PIPE)
        for c in iter(lambda: process.stdout.read(1), b""):
            sys.stdout.buffer.write(c)
        process.communicate()[0]
        if process.returncode != 0:
            logger.error("Error running command")
            sys.exit(1)
    return None


class Variables:
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._parser = argparse.ArgumentParser(
            description="Terraform deployment script.", *args, **kwargs
        )

    def add(
        self,
        name: str,
        *,
        var_type: type = None,
        help: str,
        required: bool = False,
        choices: list = [],
        example: str = "",
    ) -> None:
        example = f"Example: {example}" if len(example) > 0 else ""
        leading_dash = "--" if len(name) > 1 else "-"
        self._parser.add_argument(
            leading_dash + name,
            type=var_type,
            help=help + " Example: " + example,
            choices=choices if len(choices) > 0 else None,
            required=required,
        )

    def compile(self) -> None:
        cli_args = self._parser.parse_args()
        for key, val in vars(cli_args).items():
            setattr(self, key.replace("-", "_"), val)


def main(*, var_map: dict, template_tf_dir: str):

    tmpdir = setup_temp_dir(template_tf_dir=template_tf_dir)

    for file in os.listdir(tmpdir.name):
        if is_tf_file(file):
            for key in var_map.keys():
                replace_var(
                    filename=os.path.join(tmpdir.name, file),
                    var=key,
                    value=var_map[key],
                )
    os.chdir(tmpdir.name)
    run_tf_routine()
    tmpdir.cleanup()


if __name__ == "__main__":
    logger.info("Starting deploy script...")
    v = Variables()

    v.add("org", help="organization name.", required=True)
    v.add("env", help="environment name.", required=True)
    v.compile()
    main(
        template_tf_dir="terraform",
        var_map={
            "env": v.env,
            "org": v.org,
        },
    )