<reponame>socotra-marketplace/tap-help-scripting<filename>python-kit/src/socotra_marketplace_helpers/stubs.py
'''Utilities for building Marketplace configs from socotra-app.json'''
from typing import Tuple

import json
import os

from .jwt import sign as _sign
from .types import SAC

_domain = os.getenv('DOMAIN', 'sandbox.socotra.com')
_tenant = os.getenv('TENANT', 'marketplace-partner')

_default_hostname = os.getenv('TENANT_HOSTNAME', f'{_tenant}.co.{_domain}')
_default_base_url = os.getenv('SANDBOX_BASE_URL', f'https://api.{_domain}')

# Sandbox API URL, tenant hostname, token
def _meta() -> Tuple[str, str, str]:
    '''Returns a tuple suitable for STM construction'''
    hostname, url = _default_hostname, _default_base_url
    token = _sign(claims={'aud': [hostname]})
    jwt = os.getenv('JWT', token)
    return (url, hostname, jwt)

def _pair(data: dict, product_named: str = 'socotra-app.json') -> SAC:
    '''Returns a tuple suitable for SAC construction'''
    field_mapping = {}
    mapped = [(product_named, field_mapping)]
    vendor = {}
    join = lambda *args: '-'.join(args)
    for field in data['config']['fields']:
        name = field.get('name', '')
        if name == '':
            continue
        values = field.get('values', None)
        if field.get('type', 'mapped') == 'mapped':
            value = join(product_named, 'mapped', name)
            field_mapping[name] = value
        elif values is not None and len(values) > 0:
            first = values[0].get('value', name)
            #vendor[name] = join(product_named, 'select', first)
            vendor[name] = first
        else:
            vendor[name] = join(product_named, 'vendor', name)
    return mapped, vendor

def read_config_from(filename: str, depth=0, dirpath=os.getcwd()) -> SAC:
    '''Recursively (up to FS root) find .json file, then parse it like a SAC'''
    data = None
    path = os.path.join(dirpath, filename)
    try:
        with open(path, encoding='utf-8', mode='r') as data_source:
            data = json.load(data_source)
        if data is None or not len(data['config']['fields']) > 0:
            return None
    except (OSError, ValueError) as err:
        dirp = os.path.dirname(dirpath)
        if dirp == dirpath or depth > 100:
            raise Exception(f'No {filename} found (dirs: {depth})') from err
        return read_config_from(filename, depth=depth+1, dirpath=dirp)
    tenant_metadata = _meta()
    mapped, vendor = _pair(data)
    return SAC(tenant_metadata, mappings=mapped, settings=vendor)

'''Utilities for building Marketplace configs from socotra-app.json'''
from typing import Tuple

import json
import os

from .jwt import sign as _sign
from .types import SAC

_domain = os.getenv('DOMAIN', 'sandbox.socotra.com')
_tenant = os.getenv('TENANT', 'marketplace-partner')

_default_hostname = os.getenv('TENANT_HOSTNAME', f'{_tenant}.co.{_domain}')
_default_base_url = os.getenv('SANDBOX_BASE_URL', f'https://api.{_domain}')

# Sandbox API URL, tenant hostname, token
def _meta() -> Tuple[str, str, str]:
    '''Returns a tuple suitable for STM construction'''
    hostname, url = _default_hostname, _default_base_url
    token = _sign(claims={'aud': [hostname]})
    jwt = os.getenv('JWT', token)
    return (url, hostname, jwt)

def _pair(data: dict, product_named: str = 'socotra-app.json') -> SAC:
    '''Returns a tuple suitable for SAC construction'''
    field_mapping = {}
    mapped = [(product_named, field_mapping)]
    vendor = {}
    join = lambda *args: '-'.join(args)
    for field in data['config']['fields']:
        name = field.get('name', '')
        if name == '':
            continue
        values = field.get('values', None)
        if field.get('type', 'mapped') == 'mapped':
            value = join(product_named, 'mapped', name)
            field_mapping[name] = value
        elif values is not None and len(values) > 0:
            first = values[0].get('value', name)
            #vendor[name] = join(product_named, 'select', first)
            vendor[name] = first
        else:
            vendor[name] = join(product_named, 'vendor', name)
    return mapped, vendor

def read_config_from(filename: str, depth=0, dirpath=os.getcwd()) -> SAC:
    '''Recursively (up to FS root) find .json file, then parse it like a SAC'''
    data = None
    path = os.path.join(dirpath, filename)
    try:
        with open(path, encoding='utf-8', mode='r') as data_source:
            data = json.load(data_source)
        if data is None or not len(data['config']['fields']) > 0:
            return None
    except (OSError, ValueError) as err:
        dirp = os.path.dirname(dirpath)
        if dirp == dirpath or depth > 100:
            raise Exception(f'No {filename} found (dirs: {depth})') from err
        return read_config_from(filename, depth=depth+1, dirpath=dirp)
    tenant_metadata = _meta()
    mapped, vendor = _pair(data)
    return SAC(tenant_metadata, mappings=mapped, settings=vendor)
Utilities for building Marketplace configs from socotra-app.json # Sandbox API URL, tenant hostname, token Returns a tuple suitable for STM construction Returns a tuple suitable for SAC construction #vendor[name] = join(product_named, 'select', first) Recursively (up to FS root) find .json file, then parse it like a SAC