def query_context_modified()

in superset/security/manager.py [0:0]

• 38 lines of code
• 22 McCabe index (conditional complexity)

def query_context_modified(query_context: "QueryContext") -> bool:
    """
    Check if a query context has been modified.

    This is used to ensure guest users don't modify the payload and fetch data
    different from what was shared with them in dashboards.
    """
    form_data = query_context.form_data
    stored_chart = query_context.slice_

    # native filter requests
    if form_data is None or stored_chart is None:
        return False

    # cannot request a different chart
    if form_data.get("slice_id") != stored_chart.id:
        return True

    stored_query_context = (
        json.loads(cast(str, stored_chart.query_context))
        if stored_chart.query_context
        else None
    )

    # compare columns and metrics in form_data with stored values
    for key, equivalent in [
        ("metrics", ["metrics"]),
        ("columns", ["columns", "groupby"]),
        ("groupby", ["columns", "groupby"]),
        ("orderby", ["orderby"]),
    ]:
        requested_values = {freeze_value(value) for value in form_data.get(key) or []}
        stored_values = {
            freeze_value(value) for value in stored_chart.params_dict.get(key) or []
        }
        if not requested_values.issubset(stored_values):
            return True

        # compare queries in query_context
        queries_values = {
            freeze_value(value)
            for query in query_context.queries
            for value in getattr(query, key, []) or []
        }
        if stored_query_context:
            for query in stored_query_context.get("queries") or []:
                for key in equivalent:
                    stored_values.update(
                        {freeze_value(value) for value in query.get(key) or []}
                    )

        if not queries_values.issubset(stored_values):
            return True

    return False