automation: pull up terraform funcs to abstract class

see #1
This commit is contained in:
Iain Learmonth 2022-05-08 13:01:15 +01:00
parent 9b8ac493b1
commit 082de33b5d
7 changed files with 194 additions and 143 deletions

View file

@ -1,56 +1,30 @@
import json
import os
import subprocess
from typing import Dict, Any
import jinja2
from app import app
class BaseAutomation:
short_name = None
"""
The short name of the automation provider. This is used as an opaque token throughout
the portal system.
"""
def working_directory(self, filename=None):
def automate(self):
raise NotImplementedError()
def working_directory(self, filename=None) -> str:
"""
Provides a filesystem path that can be used during the automation run.
This is currently a persistent path, but this should not be relied upon
as future versions may use disposable temporary paths instead. State that
is needed in subsequent runs should be stored elsewhere.
:param filename: the filename inside the working directory to create a path for
:return: filesystem path for that filename
"""
return os.path.join(
app.config['TERRAFORM_DIRECTORY'],
self.short_name or self.__class__.__name__.lower(),
filename or ""
)
def write_terraform_config(self, template: str, **kwargs):
tmpl = jinja2.Template(template)
with open(self.working_directory("main.tf"), 'w') as tf:
tf.write(tmpl.render(**kwargs))
def terraform_init(self):
subprocess.run(
['terraform', 'init'],
cwd=self.working_directory())
def terraform_plan(self):
plan = subprocess.run(
['terraform', 'plan'],
cwd=self.working_directory())
def terraform_apply(self, refresh: bool = True, parallelism: int = 10):
subprocess.run(
['terraform', 'apply', f'-refresh={str(refresh).lower()}', '-auto-approve',
f'-parallelism={str(parallelism)}'],
cwd=self.working_directory())
def terraform_show(self) -> Dict[str, Any]:
terraform = subprocess.run(
['terraform', 'show', '-json'],
cwd=os.path.join(
self.working_directory()),
stdout=subprocess.PIPE)
return json.loads(terraform.stdout)
def terraform_output(self) -> Dict[str, Any]:
terraform = subprocess.run(
['terraform', 'output', '-json'],
cwd=os.path.join(
self.working_directory()),
stdout=subprocess.PIPE)
return json.loads(terraform.stdout)