(sql: str, ds: CoreDatasource | AssistantOutDsSchema)
| 717 | |
| 718 | |
| 719 | def check_sql_read(sql: str, ds: CoreDatasource | AssistantOutDsSchema): |
| 720 | try: |
| 721 | normalized_sql = sql.strip().lstrip("(").strip() |
| 722 | first_keyword = normalized_sql.split(None, 1)[0].upper() if normalized_sql else "" |
| 723 | allowed_read_commands = {"SELECT", "WITH", "SHOW", "DESCRIBE", "DESC", "EXPLAIN"} |
| 724 | denied_write_commands = { |
| 725 | "INSERT", "UPDATE", "DELETE", "CREATE", "DROP", "ALTER", |
| 726 | "TRUNCATE", "MERGE", "COPY", "REPLACE", "GRANT", "REVOKE", |
| 727 | "USE", "SET", "CALL" |
| 728 | } |
| 729 | |
| 730 | if not first_keyword: |
| 731 | raise ValueError("Parse SQL Error") |
| 732 | if first_keyword in denied_write_commands: |
| 733 | return False |
| 734 | |
| 735 | dialect = None |
| 736 | if equals_ignore_case(ds.type, 'mysql', 'doris', 'starrocks'): |
| 737 | dialect = 'mysql' |
| 738 | elif equals_ignore_case(ds.type, 'sqlServer'): |
| 739 | dialect = 'tsql' |
| 740 | elif equals_ignore_case(ds.type, 'hive'): |
| 741 | dialect = 'hive' |
| 742 | |
| 743 | statements = sqlglot.parse(sql, dialect=dialect) |
| 744 | |
| 745 | if not statements: |
| 746 | raise ValueError("Parse SQL Error") |
| 747 | |
| 748 | write_types = ( |
| 749 | exp.Insert, exp.Update, exp.Delete, |
| 750 | exp.Create, exp.Drop, exp.Alter, |
| 751 | exp.Merge, exp.Copy |
| 752 | ) |
| 753 | |
| 754 | for stmt in statements: |
| 755 | if stmt is None: |
| 756 | continue |
| 757 | if isinstance(stmt, write_types): |
| 758 | return False |
| 759 | |
| 760 | return first_keyword in allowed_read_commands |
| 761 | |
| 762 | except Exception as e: |
| 763 | raise ValueError(f"Parse SQL Error: {e}") |
| 764 | |
| 765 | |
| 766 | def checkParams(extraParams: str, illegalParams: List[str]): |
no test coverage detected