Determine if the query is dropping a specific database.
(queries: str, dbname: str | None)
| 401 | |
| 402 | |
| 403 | def is_dropping_database(queries: str, dbname: str | None) -> bool: |
| 404 | """Determine if the query is dropping a specific database.""" |
| 405 | result = False |
| 406 | if dbname is None: |
| 407 | return False |
| 408 | |
| 409 | def normalize_db_name(db: str) -> str: |
| 410 | return db.lower().strip('`"') |
| 411 | |
| 412 | dbname = normalize_db_name(dbname) |
| 413 | |
| 414 | for query in sqlparse.parse(queries): |
| 415 | keywords = [t for t in query.tokens if t.is_keyword] |
| 416 | if len(keywords) < 2: |
| 417 | continue |
| 418 | if keywords[0].normalized in ("DROP", "CREATE") and keywords[1].value.lower() in ( |
| 419 | "database", |
| 420 | "schema", |
| 421 | ): |
| 422 | database_token = next((t for t in query.tokens if isinstance(t, Identifier)), None) |
| 423 | if database_token is not None and normalize_db_name(database_token.get_name()) == dbname: |
| 424 | result = keywords[0].normalized == "DROP" |
| 425 | return result |
| 426 | |
| 427 | |
| 428 | def need_completion_refresh(queries: str) -> bool: |