SQL DDL to Kusto KQL

sql_to_kql.py

from typing import Tuple, List
# pip install sqlparse
import sqlparse

# https://docs.microsoft.com/en-us/azure/data-explorer/kusto/query/scalar-data-types/
SQL_TYPE_TO_KQL_TYPE = {
    'int': 'int',
    'varchar': 'string',
    'datetime': 'datetime',
    'bool': 'bool'
}

def read_sql() -> str:
    # can be open('/path/to/ddl.sql')
    return """
    CREATE TABLE FOO (A INT, B VARCHAR(255), C BOOL, D DATETIME);
    """

def format_column(column_name:str, column_type:str) -> str:
    # [''] is the kusto way of escaping keywords
    column_type = column_type.lower()
    if  not column_type in SQL_TYPE_TO_KQL_TYPE:
        raise NotImplementedError(f"please provide a mapping for the type '{column_type}' in SQL_TYPE_TO_KQL_TYPE")

    kql_type = SQL_TYPE_TO_KQL_TYPE[column_type]
    return f"['{column_name}']: {kql_type}"

def kql_from_columns(table: str, columns: List[Tuple[str, str]]) -> str:
    columns_as_string = ''
    for idx, col in enumerate(columns):
        columns_as_string += format_column(column_name=col[0], column_type=col[1])
        if idx != len(columns) - 1:
            columns_as_string += ' , '

    return f'.create table {table} ({columns_as_string})'

def columns_from_ast(ast) -> Tuple[str, List[Tuple[str, str]]]:
    table_name = None
    columns = []

    for t in ast.tokens:
        if t.__class__.__name__ == 'Identifier':
            table_name = t.value
        elif t.__class__.__name__ == 'Parenthesis':
            # ignore the paranthesis themselves, and split
            for col in t.value[1:-1].split(','):
                parts = col.strip().split(' ')
                name = parts[0].strip()
                # ignore type size, and the rest of the column modifiers
                ctype = parts[1].split('(')[0].strip()
                columns.append((name, ctype))

    return table_name, columns

def main():
    sql = read_sql()
    ddls = sqlparse.split(sql)
    for ddl in ddls:
        ast = sqlparse.parse(ddl)
        table_name, columns = columns_from_ast(ast[0])
        print(kql_from_columns(table_name, columns))

if __name__ == '__main__':
    main()