Check if a query is a simple single-table UPDATE.
(query: str)
| 360 | |
| 361 | # todo: handle "UPDATE LOW_PRIORITY" and "UPDATE IGNORE" |
| 362 | def query_is_single_table_update(query: str) -> bool: |
| 363 | """Check if a query is a simple single-table UPDATE.""" |
| 364 | cleaned_query_for_parsing_only = sqlparse.format(query, strip_comments=True) |
| 365 | cleaned_query_for_parsing_only = re.sub(r'\s+', ' ', cleaned_query_for_parsing_only) |
| 366 | if not cleaned_query_for_parsing_only: |
| 367 | return False |
| 368 | parsed = sqlparse.parse(cleaned_query_for_parsing_only) |
| 369 | if not parsed: |
| 370 | return False |
| 371 | statement = parsed[0] |
| 372 | try: |
| 373 | retval = bool( |
| 374 | statement[0].value.lower() == 'update' |
| 375 | and statement[1].is_whitespace |
| 376 | and ',' not in statement[2].value # multiple tables |
| 377 | and statement[3].is_whitespace |
| 378 | and statement[4].value.lower() == 'set' |
| 379 | ) |
| 380 | except IndexError: |
| 381 | retval = False |
| 382 | |
| 383 | return retval |
| 384 | |
| 385 | |
| 386 | def is_destructive(keywords: list[str], queries: str) -> bool: |
no outgoing calls