[ksql] Basic support for DESCRIBE
@@ -81,13 +81,14 @@ class KSqlApi(object):
metadata = []
if statement.strip().lower().startswith('select') or statement.strip().lower().startswith('print'):
- # STREAMS requires a LIMIT currently or https://github.com/bryanyang0528/ksql-python/pull/60
+ # STREAMS requires a LIMIT currently or will hang without https://github.com/bryanyang0528/ksql-python/pull/60
result = self.client.query(statement)
for line in ''.join(list(result)).split('\n'):
# Until https://github.com/bryanyang0528/ksql-python/issues/57
# columns = line.keys()
# data.append([line[col] for col in columns])
data.append([line])
+ # TODO: WS to plug-in
metadata = [['Row', 'STRING']]
else:
data, metadata = self._decode_result(
@@ -112,6 +113,17 @@ class KSqlApi(object):
json.dumps(line[column])
)
data.append(row)
+ elif result['@type'] == 'sourceDescription':
+ columns = ['name', 'description']
+ for key in result['sourceDescription']:
+ if key == 'fields':
+ data.append(['', ''])
+ data.append(['Fields:', ''])
+ for field in result['sourceDescription'][key]:
+ data.append([field['name'], field['schema']['type']])
+ data.append(['', ''])
+ else:
+ data.append([key, result['sourceDescription'][key]])
columns = [[col, 'STRING'] for col in columns]
