Created
March 15, 2022 21:01
-
-
Save danield137/3c1236aa17cb1c39d309a7c9cc7b5716 to your computer and use it in GitHub Desktop.
SQL DDL to Kusto KQL
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| 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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment