Pretty massive overhaul so that generated charts follow better current iterations (including making use of subcharts etc...)

This commit is contained in:
Alan Bridgeman 2025-12-30 19:01:55 -06:00
parent 826b9a7198
commit 37baf7e410
17 changed files with 2240 additions and 735 deletions

View file

@ -1,151 +1,621 @@
import json
import os, sys, json
from src.Ingress import Ingress
from src.Service import Service
from src.Dependency import Dependency
from src.Database import Database
from src.HashicorpVault import HashicorpVault
from src.MongoDB import MongoDB
from src.AzureTableStorage import AzureTableStorage
from src.Redis import Redis
from src.LoggingSidecar import LoggingSidecar
from src.OAuth import OAuth
from src.ThirdPartyService import ThirdPartyService
from src.Deployment import Deployment
from src.HelmChart import HelmChart
if __name__ == '__main__':
with open('input.json', 'r') as f:
data = json.load(f)
def print_help():
"""Print help message."""
print('Usage: python create-helm-chart.py [input_file]')
print()
print('This script generates a Helm chart based on an input JSON file (default - input.json).')
def parse_args(args: list[str]) -> dict[str, str]:
"""Parse command line arguments.
Args:
args (list[str]): List of command line arguments.
Returns:
dict[str, str]: Dictionary of results from parsing.
"""
input_file = 'input.json'
# Check if any arguments were provided
if len(args) > 1:
# Verify the correct number of arguments were provided
if len(args) == 2:
if args[1] == '--help':
# if the user requested help, display it and exit
print_help()
exit(0)
else:
# If a single argument was provided, assume it's the input file
input_file = args[1]
else:
print(f'Invalid number of arguments: {len(sys.argv)}')
print()
print_help()
exit(1)
return { 'input_file': input_file }
def get_chart_info(data: dict[str, any]) -> dict[str, any]:
"""Extract Helm chart information from input JSON data.
Note, this function also performs validation on the input data to ensure required fields are present.
And it also provides default values as needed.
Args:
data (dict[str, any]): Input JSON data.
Returns:
dict[str, any]: Extracted Helm chart information.
"""
# Verify top level 'chart' field exists
if 'chart' not in data:
raise Exception('No chart information found in input JSON file. But chart information is required to create a Helm chart.')
# --- Required Fields ---
# Verify 'name' field exists
if 'name' not in data['chart']:
raise Exception('No chart name found in input JSON file. But chart name is required to create a Helm chart.')
# The API version of the Helm chart itself
api_version = data['chart']['apiVersion']
# The version of the application that the Helm chart is deploying
app_version = data['chart']['appVersion']
# A description of the Helm chart
chart_description = data['chart']['description']
# The URL of the Helm chart's home page
chart_homepage = data['chart']['homepage']
# The maintainers of the Helm chart
maintainers = data['chart']['maintainers']
# The name of the Helm chart
chart_name = data['chart']['name']
# The sources of the Helm chart
sources = data['chart']['sources']
# The version of the Helm chart
chart_version = data['chart']['version']
# --- Required Fields with Defaults ---
# The API version of the Helm chart itself (defaults to `v2` if not specified)
# `v2` is the default because we make heavy use of dependencies, which are not supported (or supported as well) in `v1` charts
api_version = 'v2'
if 'apiVersion' in data['chart']:
api_version = data['chart']['apiVersion']
# The version of the Helm chart (defaults to `1.0.0` if not specified)
# `1.0.0` is the default because it's a reasonable starting version for a new chart
chart_version = '1.0.0'
if 'version' in data['chart']:
chart_version = data['chart']['version']
# --- Optional Fields ---
# The version of the application that the Helm chart is deploying
# Note, can be unset (variable being `None`)
app_version = None
if 'appVersion' in data['chart']:
app_version = data['chart']['appVersion']
# A description of the Helm chart
# Note, can be unset (variable being `None`)
chart_description = None
if 'description' in data['chart']:
chart_description = data['chart']['description']
# The URL of the Helm chart's home page
# Note, can be unset (variable being `None`)
chart_homepage = None
if 'homepage' in data['chart']:
chart_homepage = data['chart']['homepage']
# The maintainers of the Helm chart
# Note, can be unset (variable being `None`)
maintainers = None
if 'maintainers' in data['chart']:
maintainers = data['chart']['maintainers']
# The sources of the Helm chart
sources = None
if 'sources' in data['chart']:
sources = data['chart']['sources']
return {
'api_version': api_version,
'app_version': app_version,
'chart_description': chart_description,
'chart_homepage': chart_homepage,
'maintainers': maintainers,
'chart_name': chart_name,
'sources': sources,
'chart_version': chart_version
}
def get_image_info(data: dict[str, any]) -> dict[str, str]:
"""Extract image information from input JSON data.
Args:
data (dict[str, any]): Input JSON data.
Returns:
dict[str, str]: Extracted image information.
"""
# Verify top level 'image' field exists
if 'image' not in data:
raise Exception('No image information found in input JSON file. But image information is required to create a Helm chart.')
# Verify 'repository' field exists
if 'repository' not in data['image']:
raise Exception('No image repository found in input JSON file. But image repository is required to create a Helm chart.')
# Note, the 'repository' can be in a few different forms:
# 1. Just the registry/repository itself, specifying `name` separately
# 2. As an encompassing value, representing some combination of `<registry/repository>/<name>`
#
# Note, shouldn't use `<registry/repository>:<tag>` and specify `name` separately as it causes incorrect output and is confusing.
# But `<registry/repository>/<name>` and `<tag>` separately is perfectly valid.
#
# This multi-format nature is why it and only it is required, while `name` and `tag` are optional.
#
image_repository = data['image']['repository']
image_pull_policy = data['image']['pullPolicy']
# Append 'name' field to repository if it exists
if 'name' in data['image']:
image_repository += '/' + data['image']['name']
if 'tag' not in data['image']:
raise Exception('No image tag found in input JSON file. But image tag is required to create a Helm chart.')
image_tag = data['image']['tag']
# Append 'tag' field to repository if it exists
#if 'tag' in data['image']:
# image_repository += ':' + data['image']['tag']
# The image pull policy (defaults to `IfNotPresent` if not specified)
# `IfNotPresent` is the default because it's a reasonable default for most use cases
image_pull_policy = 'IfNotPresent'
if 'pullPolicy' in data['image']:
image_pull_policy = data['image']['pullPolicy']
return {
'image_repository': image_repository,
'image_tag': image_tag,
'image_pull_policy': image_pull_policy
}
def get_ingress_info(data: dict[str, any]) -> dict[str, str]:
"""Extract ingress information from input JSON data.
Args:
data (dict[str, any]): Input JSON data.
Returns:
dict[str, str]: Extracted ingress information.
"""
# Verify top level 'ingress' field exists
if 'ingress' not in data:
raise Exception('No ingress information found in input JSON file. But ingress information is required to create a Helm chart.')
# Verify 'hostname' field exists
if 'hostname' not in data['ingress']:
raise Exception('No ingress hostname found in input JSON file. But ingress hostname is required to create a Helm chart.')
hostname = data['ingress']['hostname']
ingress = Ingress(hostname)
service = Service()
return {
'hostname': hostname
}
templates = [ingress, service]
def get_db_info(data: dict[str, any]) -> tuple[bool, dict[str, Database | Dependency | None]]:
"""Extract database information from input JSON data.
Args:
data (dict[str, any]): Input JSON data.
Returns:
tuple[bool, dict[str, Database | Dependency]]: A tuple where the first element indicates whether a database is used,
and the second element is a dictionary containing the database object and the dependency object.
"""
uses_db = False
uses_secrets_vault = False
nosql = None
uses_cache = False
third_party_services = []
extra_env_vars = {}
db = None
db_dependency = None
if 'db' in data and data['db'] != False:
db_name = data['db']['name']
db_host = data['db']['host']
db_user = data['db']['user']
db_password = data['db']['password']
if 'name' not in data['db']:
raise Exception('No database name found in input JSON file. But database name is required to create the Helm chart with database support.')
db = Database(db_name, db_host, db_user, db_password)
db_name = data['db']['name']
if 'host' not in data['db']:
raise Exception('No database host found in input JSON file. But database host is required to create the Helm chart with database support.')
db_host = data['db']['host']
if 'user' not in data['db']:
raise Exception('No database user found in input JSON file. But database user is required to create the Helm chart with database support.')
db_user = data['db']['user']
if 'password' not in data['db']:
raise Exception('No database password found in input JSON file. But database password is required to create the Helm chart with database support.')
db_password = data['db']['password']
uses_db = True
templates.append(db)
db = Database(db_name, db_host, db_user, db_password)
db_dependency = Dependency('db-deploy', '1.0.2', 'https://helm.bridgemanaccessible.ca/', 'database', 'database.enabled')
return uses_db, { 'db': db, 'db_dependency': db_dependency }
def get_vault_info(data: dict[str, any]) -> tuple[bool, dict[str, HashicorpVault | Dependency | None]]:
"""Extract HashiCorp Vault information from input JSON data.
Args:
data (dict[str, any]): Input JSON data.
Returns:
tuple[bool, dict[str, HashicorpVault | Dependency]]: A tuple where the first element indicates whether Vault is used,
and the second element is a dictionary containing the Vault object and the dependency object.
"""
uses_vault = False
vault = None
vault_dependency = None
if 'vault' in data and data['vault'] != False:
if 'image' not in data['vault']:
raise Exception('No Vault image information found in input JSON file. But Vault image information is required to create the Helm chart with Vault support.')
if 'repository' not in data['vault']['image'] or 'tag' not in data['vault']['image']:
raise Exception('Incomplete Vault image information found in input JSON file. Both repository and tag are required to create the Helm chart with Vault support.')
vault_image = {
'repository': data['vault']['image']['repository'],
'tag': data['vault']['image']['tag']
}
if 'hostname' not in data['vault']:
raise Exception('No Vault hostname found in input JSON file. But Vault hostname is required to create the Helm chart with Vault support.')
vault_hostname = data['vault']['hostname']
vault_storage_class = data['vault']['storageClass']
#vault_storage_class = data['vault']['storageClass']
if 'policyCapabilities' not in data['vault']:
raise Exception('No Vault policy capabilities found in input JSON file. But Vault policy capabilities are required to create the Helm chart with Vault support.')
vault = HashicorpVault(image=vault_image, hostname=vault_hostname, storage_class=vault_storage_class)
vault_policy_capabilities = data['vault']['policyCapabilities']
uses_vault = True
uses_secrets_vault = True
templates.append(vault)
vault = HashicorpVault(image=vault_image, hostname=vault_hostname, policy_capabilities=vault_policy_capabilities)
vault_dependency = Dependency('ba-custom-hashicorp-vault', '1.0.6', 'https://helm.bridgemanaccessible.ca/', 'vault', 'vault.enabled')
return uses_vault, { 'vault': vault, 'vault_dependency': vault_dependency }
def get_nosql_info(data: dict[str, any]) -> tuple[bool, dict[str, MongoDB | AzureTableStorage | Dependency | None]]:
"""Extract NoSQL information from input JSON data.
Args:
data (dict[str, any]): Input JSON data.
Returns:
tuple[bool, Database | None]: A tuple where the first element indicates whether a NoSQL database is used,
and the second element is the NoSQL database object if used, otherwise None.
"""
nosql = None
nosql_dependency = None
if 'nosql' in data and data['nosql'] != False:
# Default to MongoDB
nosql_type = 'mongodb'
if 'type' in data['nosql']:
nosql_type = data['nosql']['type']
if 'dbName' not in data['nosql']:
raise Exception('No NoSQL database name found in input JSON file. But NoSQL database name is required to create the Helm chart with NoSQL database support.')
nosql_db_name = data['nosql']['dbName']
nosql_user = data['nosql']['user']
nosql_password = data['nosql']['password']
tables = data['nosql']['tables']
if 'groupings' not in data['nosql']:
raise Exception('No NoSQL groupings found in input JSON file. But NoSQL groupings are required to create the Helm chart with NoSQL database support.')
mongo = MongoDB(nosql_db_name, nosql_user, nosql_password, tables)
groupings = data['nosql']['groupings']
nosql = mongo
if nosql_type == 'mongodb':
if 'user' not in data['nosql']:
raise Exception('No NoSQL database user found in input JSON file. But NoSQL database user is required to create the Helm chart with NoSQL (MongoDB) database support.')
nosql_user = data['nosql']['user']
if 'password' not in data['nosql']:
raise Exception('No NoSQL database password found in input JSON file. But NoSQL database password is required to create the Helm chart with NoSQL (MongoDB) database support.')
templates.append(mongo)
nosql_password = data['nosql']['password']
mongo = MongoDB(nosql_db_name, nosql_user, nosql_password, groupings)
nosql = mongo
elif nosql_type == 'az_table_storage':
nosql_key = data['nosql']['key']
az_table_storage = AzureTableStorage(nosql_db_name, nosql_key, groupings)
nosql = az_table_storage
nosql_dependency = Dependency('nosql-deploy', '1.0.5', 'https://helm.bridgemanaccessible.ca/', 'nosql', 'nosql.enabled')
return nosql is not None, { 'nosql': nosql, 'nosql_dependency': nosql_dependency }
def get_cache_info(data: dict[str, any]) -> tuple[bool, dict[str, Redis | Dependency | None]]:
"""Extract cache information from input JSON data.
Args:
data (dict[str, any]): Input JSON data.
Returns:
tuple[bool, dict[str, Redis | Dependency]]: A tuple where the first element indicates whether a cache is used,
and the second element is a dictionary containing the cache object and the dependency object.
"""
uses_cache = False
cache = None
cache_dependency = None
if 'cache' in data and data['cache'] != False:
cache_password = data['cache']['password']
if 'password' not in data['cache']:
raise Exception('No cache password found in input JSON file. But cache password is required to create the Helm chart with cache support.')
redis = Redis(cache_password)
cache_password = data['cache']['password']
uses_cache = True
templates.append(redis)
cache = Redis(cache_password)
cache_dependency = Dependency('cache-deploy', '1.0.7', 'https://helm.bridgemanaccessible.ca/', 'cache', 'cache.enabled')
return uses_cache, { 'cache': cache, 'cache_dependency': cache_dependency }
def get_oauth_info(data: dict[str, any]) -> tuple[bool, dict[str, OAuth | None]]:
"""Extract OAuth information from input JSON data.
Args:
data (dict[str, any]): Input JSON data.
Returns:
tuple[bool, dict[str, OAuth | None]]: A tuple where the first element indicates whether OAuth is used,
and the second element is a dictionary containing the OAuth object if used, otherwise None.
"""
uses_oauth = False
oauth = None
if 'oauth' in data and data['oauth'] != False:
if 'baseAppUrl' not in data['oauth']:
raise Exception('No OAuth base application URL found in input JSON file. But OAuth base application URL is required to create the Helm chart with OAuth support.')
base_app_url = data['oauth']['baseAppUrl']
if 'appAbbreviation' not in data['oauth']:
raise Exception('No OAuth application abbreviation found in input JSON file. But OAuth application abbreviation is required to create the Helm chart with OAuth support.')
app_abbreviation = data['oauth']['appAbbreviation']
if 'appName' not in data['oauth']:
raise Exception('No OAuth application name found in input JSON file. But OAuth application name is required to create the Helm chart with OAuth support.')
app_name = data['oauth']['appName']
if 'serviceName' not in data['oauth']:
raise Exception('No OAuth service name found in input JSON file. But OAuth service name is required to create the Helm chart with OAuth support.')
service_name = data['oauth']['serviceName']
if 'devPort' not in data['oauth']:
raise Exception('No OAuth development port found in input JSON file. But OAuth development port is required to create the Helm chart with OAuth support.')
dev_port = data['oauth']['devPort']
oauth = OAuth(base_app_url, app_abbreviation, app_name, service_name, dev_port)
if 'appRegContactEmail' not in data['oauth']:
raise Exception('No OAuth application registration contact email found in input JSON file. But OAuth application registration contact email is required to create the Helm chart with OAuth support.')
templates.append(oauth)
app_reg_contact_email = data['oauth']['appRegContactEmail']
uses_oauth = True
oauth = OAuth(base_app_url, app_abbreviation, app_name, service_name, dev_port, app_reg_contact_email)
return uses_oauth, { 'oauth': oauth }
def get_logging_info(data: dict[str, any]) -> tuple[bool, dict[str, LoggingSidecar | Dependency | None]]:
"""Extract logging information from input JSON data.
Args:
data (dict[str, any]): Input JSON data.
Returns:
tuple[bool, dict[str, LoggingSidecar | Dependency | None]]: A tuple where the first element indicates whether logging is used,
and the second element is a dictionary containing the Logging object if used, otherwise None.
"""
uses_logging = False
logging = None
logging_dependency = None
if 'logging' in data and data['logging'] != False:
uses_logging = True
log_aggregator_username = data['logging']['username']
log_aggregator_password = data['logging']['password']
logging = LoggingSidecar(log_aggregator_username, log_aggregator_password)
logging_dependency = Dependency('ba-logging-sidecar', '1.0.2', 'https://helm.bridgemanaccessible.ca/', 'loggingSidecar', 'loggingSidecar.enabled')
return uses_logging, { 'logging': logging, 'logging_dependency': logging_dependency }
def get_third_party_services_info(data: dict[str, any]) -> list[ThirdPartyService]:
"""Extract third-party services information from input JSON data.
Args:
data (dict[str, any]): Input JSON data.
Returns:
list[ThirdPartyService]: Extracted third-party services.
"""
third_party_services = []
if 'thirdPartyServices' in data:
if 'openai' in data['thirdPartyServices']:
openai_api_key = data['thirdPartyServices']['openai']['apiKey']
for service_name in data['thirdPartyServices'].keys():
service_vars = {}
openai = ThirdPartyService('openai', False, api_key=openai_api_key)
for key in data['thirdPartyServices'][service_name].keys():
# If the key is already in snake_case, skip it
if '_' not in key:
# Convert from camelCase to snake_case
new_key = ''
for i, c in enumerate(key):
if c.isupper() and i != 0:
new_key += '_' + c.lower()
else:
new_key += c.lower()
# Add the snake_case key to the dictionary with the value of the camelCase key
service_vars[new_key] = data['thirdPartyServices'][service_name][key]
third_party_services.append(openai)
service = ThirdPartyService(service_name, True, **service_vars) #merchant_id=moneris_merchant_id, store_id=moneris_store_id, ht_profile_id=moneris_ht_profile_id, test_merchant_id=moneris_test_merchant_id, test_store_id=moneris_test_store_id, test_ht_profile_id=moneris_test_ht_profile_id)
templates.append(openai)
if 'stripe' in data['thirdPartyServices']:
stripe_public_key = data['thirdPartyServices']['stripe']['publicKey']
stripe_secret_key = data['thirdPartyServices']['stripe']['secretKey']
stripe_test_public_key = data['thirdPartyServices']['stripe']['testPublicKey']
stripe_test_secret_key = data['thirdPartyServices']['stripe']['testSecretKey']
third_party_services.append(service)
return third_party_services
stripe = ThirdPartyService('stripe', True, public_key=stripe_public_key, secret_key=stripe_secret_key, test_public_key=stripe_test_public_key, test_secret_key=stripe_test_secret_key)
def get_extra_env_vars(data: dict[str, any]) -> dict[str, str]:
"""Extract extra environment variables from input JSON data.
Args:
data (dict[str, any]): Input JSON data.
Returns:
dict[str, str]: Extracted extra environment variables.
"""
third_party_services.append(stripe)
templates.append(stripe)
extra_env_vars = {}
if 'extraEnvVars' in data:
extra_env_vars = data['extraEnvVars']
for key, value in extra_env_vars.items():
if not isinstance(value, dict) and value.find("'") != -1:
extra_env_vars[key] = value.replace("'", '"')
deployment = Deployment(image_repository, image_pull_policy=image_pull_policy, uses_db=uses_db, uses_secrets_vault=uses_secrets_vault, nosql=nosql, uses_cache=uses_cache, third_party_services=third_party_services, **extra_env_vars)
return extra_env_vars
if __name__ == '__main__':
# Parse command line arguments
command_line_parsing_results = parse_args(sys.argv)
input_file = command_line_parsing_results['input_file']
# Verify the input file exists
if not os.path.exists(input_file):
print(f'{input_file} file not found. Please create/specify a valid {input_file} (should be based on the input.example.json file).')
exit(1)
# Load the input JSON file
# Note, we ASSUME the JSON file is valid JSON (we have no error handling for invalid JSON here)
with open(input_file, 'r') as f:
data = json.load(f)
# Extract Helm chart information from input JSON data
chart_data = get_chart_info(data)
# Extract image information from input JSON data
image_data = get_image_info(data)
# Extract ingress information from input JSON data
ingress_data = get_ingress_info(data)
ingress = Ingress(ingress_data['hostname'])
service = Service()
templates = [ingress, service]
dependencies = []
uses_db, db_objs = get_db_info(data)
uses_secrets_vault, vault_objs = get_vault_info(data)
uses_nosql, nosql_objs = get_nosql_info(data)
uses_cache, cache_objs = get_cache_info(data)
uses_oauth, oauth_objs = get_oauth_info(data)
uses_logging, logging_objs = get_logging_info(data)
third_party_services = get_third_party_services_info(data)
extra_env_vars = get_extra_env_vars(data)
if uses_db:
templates.append(db_objs['db'])
dependencies.append(db_objs['db_dependency'])
if uses_secrets_vault:
templates.append(vault_objs['vault'])
dependencies.append(vault_objs['vault_dependency'])
if uses_nosql:
templates.append(nosql_objs['nosql'])
dependencies.append(nosql_objs['nosql_dependency'])
if uses_cache:
templates.append(cache_objs['cache'])
dependencies.append(cache_objs['cache_dependency'])
if uses_oauth:
templates.append(oauth_objs['oauth'])
if uses_logging:
templates.append(logging_objs['logging'])
dependencies.append(logging_objs['logging_dependency'])
if len(third_party_services) > 0:
for third_party_service in third_party_services:
templates.append(third_party_service)
deployment = Deployment(
image_data['image_repository'],
image_data['image_tag'],
port=data['port'] if 'port' in data else 8080,
image_pull_policy=image_data['image_pull_policy'],
uses_oauth=uses_oauth,
uses_db=uses_db,
uses_secrets_vault=uses_secrets_vault,
nosql=nosql_objs['nosql'],
uses_cache=uses_cache,
third_party_services=third_party_services,
**extra_env_vars
)
templates.append(deployment)
#templates = [ingress, service, db, vault, mongo, redis, oauth, deployment, stripe, openai]
helmChart = HelmChart(chart_name, chart_description, maintainers, chart_homepage, sources, app_version, chart_version, api_version, *templates)
helmChart = HelmChart(
chart_data['chart_name'],
chart_data['chart_version'],
chart_data['api_version'],
chart_data['app_version'],
chart_data['chart_description'],
chart_data['maintainers'],
chart_data['chart_homepage'],
chart_data['sources'],
dependencies,
*templates
)
helmChart.create_templates_folder()
helmChart.write_yaml()
helmChart.write_values_yaml()
helmChart.write_filled_in_values_yaml()
helmChart.write_helmignore()
try: