Skip to content

Instantly share code, notes, and snippets.

@phaer
Created August 15, 2018 03:47
Show Gist options
  • Save phaer/46420c56c26d69b2a6c770a8e9d11959 to your computer and use it in GitHub Desktop.
Save phaer/46420c56c26d69b2a6c770a8e9d11959 to your computer and use it in GitHub Desktop.
kubernetes open api to hcl2 spec
import sys
import json
from collections import OrderedDict
from contextlib import contextmanager
def resolve_json_pointer(spec, reference):
prefix, definitions, name = reference.split('/')
return spec.get(definitions).get(name)
def dereference_json_pointer(spec, obj):
if isinstance(obj, dict):
reference = obj.get('$ref')
if reference:
obj = resolve_json_pointer(spec, reference)
for key, value in obj.items():
obj[key] = dereference_json_pointer(spec, value)
elif isinstance(obj, list):
for value in obj:
value = dereference_json_pointer(spec, value)
return obj
class HCL:
def __init__(self, indent=0):
self.indent = indent
self.lines = []
@property
def content(self):
for indent, line in self.lines:
yield ' ' * indent + line
def write(self, filename):
with open(filename, 'w') as f:
for line in content:
f.write(line)
def add(self, line, newline=True):
if newline:
line += '\n'
self.lines.append((self.indent, line))
def add_comment(self, comment):
for line in comment.split('\n'):
self.add('# {}'.format(line))
def add_attribute(self, name, value, required=False):
with self.with_block('attr "{}"'.format(name)):
self.add('type = {}'.format(value))
if required:
self.add('required = True')
def start_block(self, label=None):
self.add(label + ' {' if label else '{')
self.indent += 2
def end_block(self):
self.indent -= 2
self.add('}')
@contextmanager
def with_block(self, label):
self.start_block(label)
yield
self.end_block()
def add_definition(self, definition, label=None):
if label:
with self.with_block(label):
self._add_definition(definition)
else:
self._add_definition(definition)
def _add_definition(self, definition):
with self.with_block('object'):
for name, prop in definition['properties'].items():
self.add_comment(prop.get('description', 'no description'))
properties = prop.get('properties')
if properties:
self.add_definition(prop, label='block "{}"'.format(name))
else:
typ = self._resolve_type(prop['type'])
if typ == 'array':
self._handle_complex_type('items', 'list', name, prop)
elif typ == 'object':
self._handle_complex_type('additionalProperties', 'map', name, prop)
else:
self.add_attribute(name, typ)
self.add('')
def _resolve_type(self, typ):
types = {
'integer': 'number',
'boolean': 'bool'
}
return types[typ] if typ in types else typ
def _handle_complex_type(self, key, kind, name, prop):
typ = self._resolve_type(prop[key].get('type'))
if typ:
self.add_attribute(name, '{}({})'.format(kind, typ))
else:
if kind == 'list':
with self.with_block('block_list "{}"'.format(name)):
self.add('block_type = "{}"'.format(self._singular(name)))
self.add_definition(prop[key])
else:
with self.with_block('block "{}"'.format(name)):
self.add_definition(prop[key])
def _singular(self, s):
return s[:-1] if s.endswith('s') else s
if __name__ == '__main__':
apis = [
'io.k8s.kubernetes.pkg.api.v1',
'io.k8s.kubernetes.pkg.apis.extensions.v1beta1'
]
with open('openapi.json') as f:
spec = json.load(f, object_pairs_hook=OrderedDict)
for full_name, definition in spec['definitions'].items():
api, name = full_name.rsplit('.', 1)
if api in apis:
hcl = HCL()
path = 'spec/{}.hcl'.format(name)
obj = resolve_json_pointer(spec, '#/definitions/{}'.format(full_name))
definition = dereference_json_pointer(spec, obj)
hcl.add_definition(definition)
with open(path, 'w') as f:
f.write(''.join(hcl.content))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment