publish.py

import os
import requests
import json
from ruamel.yaml import YAML
from pathlib import Path
from io import StringIO

def yaml() -> YAML:
   yml = YAML()
   yml.indent(mapping=2, sequence=4, offset=2)
   yml.allow_unicode = True
   yml.default_flow_style = False
   yml.preserve_quotes = True
   return yml

def update_addon_file_values_schema(ADDON_FILE, VALUES_SCHEMA_JSON_FILE):
    # Load the VALUES SCHEMA JSON file content
    with open(VALUES_SCHEMA_JSON_FILE, 'r') as f:
        json_schema = json.load(f)
    properties = json_schema.get('properties', {})

    # Load the ADDON manifest
    with open(ADDON_FILE, 'r') as f:
        manifest = safe_load(f.read())

    # Update the valuesSchema field
    if not manifest or not isinstance(manifest, dict) or not manifest.get('spec'):
        print('❌ ADDON file "spec" field is missing or empty!')
        exit(1)

    manifest['spec']['valuesSchema'] = properties

    # Write back to the ADDON file
    with open(ADDON_FILE, 'w') as f:
        yml = yaml()
        yml.dump(manifest, f)

    print(f"Updated valuesSchema in {ADDON_FILE}")
    if VERBOSE:
        print("🕵️  ADDON FILE CONTENT WITH VALUE SCHEMA:", manifest)

def safe_load(content: str) -> dict:
   yml = yaml()
   return yml.load(StringIO(content))

def get_environment_urls(CLIENT_REALM, ORGANIZATION_ID):
    return {
        "auth": f"https://idm.stackspot.com/{CLIENT_REALM}/oidc/oauth/token",
        "publish": f"https://cloud-platform-horizon.stackspot.com/v1/organizations/{ORGANIZATION_ID}/addons"
    }


def authentication(CLIENT_REALM, CLIENT_ID, CLIENT_KEY, ORGANIZATION_ID):
    urls = get_environment_urls(CLIENT_REALM, ORGANIZATION_ID)
    iam_url = urls["auth"]
    iam_headers = {'Content-Type': 'application/x-www-form-urlencoded'}
    iam_data = {
        "client_id": CLIENT_ID,
        "grant_type": "client_credentials",
        "client_secret": CLIENT_KEY
    }
    print(f"⚙️  Authenticating in {CLIENT_REALM}...")
    response = requests.post(url=iam_url, headers=iam_headers, data=iam_data)
    if response.status_code == 200:
        print("✅ Authentication successful")
        return response.json().get("access_token")
    print("❌ Authentication error")
    print(f"Status: {response.status_code}, Error: {response.text}")
    exit(1)

def publish(addon_name, addon_version, publish_headers, yaml_file_path, CLIENT_REALM, VERBOSE, ORGANIZATION_ID):
    urls = get_environment_urls(CLIENT_REALM, ORGANIZATION_ID)
    publish_url = urls["publish"]
    print(f'⚙️  Publishing "{addon_name}", version "{addon_version}" to {CLIENT_REALM} organization')
    if VERBOSE:
        print(f'⚙️  PUBLICATION URL: {publish_url}')
  
    
    with open(yaml_file_path, 'r') as file:
        yaml_content = file.read()
    
    headers = {
        "Authorization": publish_headers["Authorization"],
        "Content-Type": "application/yaml",
        "Accept": "application/json"
    }

    if VERBOSE:
        print(f'⚙️  YAML: {yaml_content}')

    response = requests.post(
        url=publish_url,
        headers=headers,
        data=yaml_content
    )
    
    if response.status_code == 200:
        print(f'✅ Addon "{addon_name}" published successfully (ID: {response.json().get('metadata').get('id')})')
        if VERBOSE:
            print("🕵️  ADDON RESPONSE DATA:", response.json())
    else:
        print(f'❌ Addon "{addon_name}" publication failed!')
        print(f"Status: {response.status_code}, Error: {response.reason}")
        if VERBOSE:
            print("🕵️  ERROR RESPONSE DATA:", response.text)
        exit(1)


# Environment variables
CLIENT_ID = os.getenv("CLIENT_ID")
CLIENT_KEY = os.getenv("CLIENT_KEY")
CLIENT_REALM = os.getenv("CLIENT_REALM")
VERBOSE = os.getenv("VERBOSE")
ADDON_FILE = os.getenv("ADDON_FILE")
VALUES_SCHEMA_JSON_FILE = os.getenv("VALUES_SCHEMA_JSON_FILE")

if VALUES_SCHEMA_JSON_FILE is not None:
    update_addon_file_values_schema(ADDON_FILE, VALUES_SCHEMA_JSON_FILE)

if not all([CLIENT_ID, CLIENT_KEY, CLIENT_REALM, ADDON_FILE]):
    print("❌  Missing required environment variables!")
    exit(1)

# Load the YAML file to extract metadata for logging purposes
with open(Path(ADDON_FILE), 'r') as file:
    yaml_data = safe_load(file.read())

# Extract data from YAML
metadata = yaml_data.get('metadata', {})
spec = yaml_data.get('spec', {})

addon_name = metadata.get('name')
addon_version = metadata.get('version')
helm = spec.get('helm')
organization_id = spec.get('organizationId')

required_fields = {
    "organization_id": organization_id,
    "addon_name": addon_name,
    "addon_version": addon_version,
    "helm": helm,    
}

missing_fields = [field for field, value in required_fields.items() if not value]

if missing_fields:
    print("❌  Missing addon required fields in the YAML file:")
    for field in missing_fields:
        print(f"- {field}")
    exit(1)

# Authenticate
access_token = authentication(CLIENT_REALM, CLIENT_ID, CLIENT_KEY, organization_id)
publish_headers = {"Authorization": f"Bearer {access_token}"}

# Start publishing addon by sending the YAML file directly
publish(addon_name, addon_version, publish_headers, ADDON_FILE, CLIENT_REALM, VERBOSE, organization_id)