def extract_signature()

in elasticapm/instrumentation/packages/dbapi2.py [0:0]

• 36 lines of code
• 12 McCabe index (conditional complexity)

def extract_signature(sql):
    """
    Extracts a minimal signature from a given SQL query
    :param sql: the SQL statement
    :return: a string representing the signature
    """
    sql = force_text(sql)
    sql = sql.strip()
    first_space = sql.find(" ")
    if first_space < 0:
        return sql

    second_space = sql.find(" ", first_space + 1)

    sql_type = sql[0:first_space].upper()

    if sql_type in ["INSERT", "DELETE"]:
        keyword = "INTO" if sql_type == "INSERT" else "FROM"
        sql_type = sql_type + " " + keyword

        object_name = look_for_table(sql, keyword)
    elif sql_type in ["CREATE", "DROP"]:
        # 2nd word is part of SQL type
        sql_type = sql_type + sql[first_space:second_space]
        object_name = ""
    elif sql_type == "UPDATE":
        object_name = look_for_table(sql, "UPDATE")
    elif sql_type == "SELECT":
        # Name is first table
        try:
            sql_type = "SELECT FROM"
            object_name = look_for_table(sql, "FROM")
        except Exception:
            object_name = ""
    elif sql_type in ["EXEC", "EXECUTE"]:
        sql_type = "EXECUTE"
        end = second_space if second_space > first_space else len(sql)
        object_name = sql[first_space + 1 : end]
    elif sql_type == "CALL":
        first_paren = sql.find("(", first_space)
        end = first_paren if first_paren > first_space else len(sql)
        procedure_name = sql[first_space + 1 : end].rstrip(";")
        object_name = procedure_name + "()"
    else:
        # No name
        object_name = ""

    signature = " ".join(filter(bool, [sql_type, object_name]))
    return signature