in polardb-postgresql-mcp-server/server.py [0:0]
def get_sql_operation_type(sql):
"""
get sql operation type
:param sql: input sql
:return: return sql operation type ('INSERT', 'DELETE', 'UPDATE', 'DDL', or 'OTHER')
"""
parsed = sqlparse.parse(sql)
if not parsed:
return 'OTHER' #parse sql failed
# get first statement
statement = parsed[0]
# get first keyword
first_token = statement.token_first(skip_ws=True, skip_cm=True)
if not first_token:
return 'OTHER'
keyword = first_token.value.upper() # convert to upper case for uniform comparison
# judge sql type
if keyword == 'INSERT':
return 'INSERT'
elif keyword == 'DELETE':
return 'DELETE'
elif keyword == 'UPDATE':
return 'UPDATE'
elif keyword in ('CREATE', 'ALTER', 'DROP', 'TRUNCATE'):
return 'DDL'
else:
return 'OTHER'