get_config.py

# Requires
# OPM
# podman or docker, skopeo

import os
import re
import time
import json
import random
import sqlite3
import subprocess

CONTAINER_TOOL="podman"

def run_cmd(cmd, params=None, strict=True, err_msg=None):
    if not params:
        params = {}
    params.setdefault('universal_newlines', True)
    params.setdefault('encoding', 'utf-8')
    params.setdefault('stderr', subprocess.PIPE)
    params.setdefault('stdout', subprocess.PIPE)

    response = subprocess.run(cmd, **params)
    if strict and response.returncode != 0:
        raise RuntimeError(err_msg, " : ", response.stderr)
    return response.stdout


def _copy_files_from_image(image, src_path, dest_path):
    run_cmd([CONTAINER_TOOL, 'pull', image], err_msg="error pulling image " + image)
    container_command = 'unused'
    container_id = run_cmd([CONTAINER_TOOL, 'create', image, container_command])
    run_cmd([CONTAINER_TOOL, 'cp', f'{container_id.strip()}:{src_path}', dest_path])


def skopeo_inspect(*args):
    cmd = ['skopeo', 'inspect'] + list(args)
    output = run_cmd(cmd, err_msg="error skopeo inspecting image ")
    return json.loads(output)


def get_image_label(pull_spec, label):
    if pull_spec.startswith('docker://'):
        full_pull_spec = pull_spec
    else:
        full_pull_spec = f'docker://{pull_spec}'
    return skopeo_inspect(full_pull_spec, '--config').get('config', {}).get('Labels', {}).get(label)


def _get_index_database(from_index, base_dir):
    db_path = get_image_label(from_index, 'operators.operatorframework.io.index.configs.v1')
    _copy_files_from_image(from_index, db_path, base_dir)
    return os.path.join(base_dir, os.path.basename(db_path))


def get_config(from_index, base_dir):
    _ = _get_index_database(from_index, base_dir)
    return


# Change this variable for other indexes
INDEX_IMAGE="docker.io/anik120/redhat-operators:main"
BASE_DIR="./"

get_config(INDEX_IMAGE, BASE_DIR)