dhall-kubernetes/convert.py

#!/usr/bin/env python3

import requests
import re

kubernetes_tag = 'v1.11.0'
url = \
    'https://raw.githubusercontent.com/kubernetes/kubernetes/{tag}/api/openapi-spec/swagger.json' \
    .format(tag=kubernetes_tag)

# See https://kubernetes.io/docs/concepts/overview/working-with-objects/kubernetes-objects/#required-fields
# because k8s API allows PUTS etc with partial data, it's not clear from the data types OR the API which
# fields are required for A POST... so we resort to .. RTFM
always_required = {'apiVersion', 'kind', 'metadata'}


required_for = {
    'io.k8s.apimachinery.pkg.apis.meta.v1.ObjectMeta': {'name'},
}


def schema_path_from_ref(prefix, ref):
    return '{}/{}.dhall'.format(prefix, ref.split('/')[2])

def required_properties (schema_name, schema):
    required = set(schema.get('required', [])) | always_required
    if schema_name in required_for.keys():
        required |= required_for[schema_name]
    return required


def build_type(schema, path_prefix, schema_name=None):
    """
    Take an OpenAPI Schema Object and return a corresponding Dhall type.

    If the schema is a reference we translate the reference path to a
    Dhall path and return that path. The ``path_prefix`` argument is
    prepended to the returned path.
    """
    if '$ref' in schema:
        return schema_path_from_ref(path_prefix, schema['$ref'])
    elif 'type' in schema:
        typ = schema['type']
        if typ == 'object':
            return '(List {mapKey : Text, mapValue : Text})'
        elif typ == 'array':
            return 'List {}'.format(build_type(schema['items'], path_prefix))
        # Fix for the funny format parameters they use in Kube
        elif typ == 'string' and 'format' in schema and schema['format'] == 'int-or-string':
            return '< Int : Natural | String : Text >'
        else:
            return {
                'string' : 'Text',
                'boolean': 'Bool',
                'integer': 'Natural',
                'number': 'Double',
            }[typ]
    elif 'properties' in schema:
        required = required_properties(schema_name, schema)
        fields = []
        for propName, propSpec in schema['properties'].items():
            propType = build_type(propSpec, path_prefix)
            if propName not in required:
                propType = 'Optional ({})'.format(propType)
            fields.append(' {} : ({})\n'.format(labelize(propName), propType))
        return '{' + ','.join(fields) + '}'
    else:
        # There are empty schemas that only have a description.
        return '{}'


def get_default(prop, required, value):
    typ = build_type(prop, '../types')
    if not required:
        return '([] : Optional ({}))'.format(typ)
    elif value:
        return '("{}" : {})'.format(value, typ)
    else:
        raise ValueError('Missing value for required property')


def get_static_data(modelSpec):
    """
    Return a dictionary of static values that all objects of this model
    have.

    This applies only to kubernetes resources where ``kind`` and
    ``apiVersion`` are statically determined by the resource. See the
    `Kubernetes OpenAPI Spec Readme`__.

    For example for a v1 Deployment we return

    ::
        {
          'kind': 'Deployment',
          'apiVersion': 'apps/v1'
        }

    .. __: https://github.com/kubernetes/kubernetes/blob/master/api/openapi-spec/README.md#x-kubernetes-group-version-kind

    """
    if 'x-kubernetes-group-version-kind' in modelSpec:
        values = modelSpec['x-kubernetes-group-version-kind']
        if len(values) == 1:
            group = values[0].get('group', '')
            if group:
                group = group + '/'
            return {
                'kind': values[0]['kind'],
                'apiVersion': group + values[0]['version']
            }
        else:
            return {}
    else:
        return {}


def labelize(propName):
    """
    If a propName doesn't match the 'simple-label' grammar, we return a quoted label
    See: https://github.com/dhall-lang/dhall-lang/blob/1d2912067658fdbbc17696fc86f057d6f91712b9/standard/dhall.abnf#L125
    """
    if not re.match("^[a-zA-Z_][a-zA-Z0-9_\-/]*$", propName):
        return "`" + propName + "`"
    else:
        return propName


def main():
    spec = requests.get(url).json()

    for modelName, modelSpec in spec['definitions'].items():
        with open('types/' + modelName + '.dhall', 'w') as f:
            f.write('{}\n'.format(build_type(modelSpec, '.', modelName)))
        with open('default/' + modelName + '.dhall', 'w') as f:
            if 'type' in modelSpec:
                typ = build_type(modelSpec, '../types')
                # In case we have a union, we make the constructors for it
                if typ[0] == '<':
                    f.write('constructors {}\n'.format(typ))
                # Otherwise we just output the identity
                else:
                    f.write('\(a : {}) -> a\n'.format(typ))
            elif '$ref' in modelSpec:
                path = schema_path_from_ref('.', modelSpec['$ref'])
                f.write('{}\n'.format(path))
            else:
                required = required_properties(modelName, modelSpec)
                if modelName in required_for.keys():
                    required |= required_for[modelName]

                properties = modelSpec.get('properties', {})

                resource_data = get_static_data(modelSpec)
                param_names = required - set(resource_data.keys())

                # If there's multiple required props, we make it a lambda
                requiredProps = [k for k in properties if k in required]
                if len(requiredProps) > 0:
                    params = ['{} : ({})'.format(labelize(propName), build_type(propVal, '../types'))
                              for propName, propVal in properties.items()
                              if propName in param_names]
                    f.write('\(_params : {' + ', '.join(params) + '}) ->\n')

                # If it's required we're passing it in as a parameter
                KVs = [(propName, "_params." + propName)
                       if propName in param_names
                       else (propName, get_default(propDef, propName in required, resource_data.get(propName, None)))
                       for propName, propDef in properties.items()]

                # If there's no fields, should be an empty record
                if len(KVs) > 0:
                    formatted = [" {} = {}\n".format(labelize(k), v) for k, v in KVs]
                else:
                    formatted = '='
                f.write('{' + ','.join(formatted) + '} : ../types/' + modelName + '.dhall\n')

if __name__ == '__main__':
    main()