Skip to content

jupyter

jupyter

generate_search_dataflows_to_jupyter_workspaces_body 

generate_search_dataflows_to_jupyter_workspaces_body(
    filter_body: dict | None = None,
    dataflow_id: int | None = None,
) -> dict

Ensure the DATA_FLOW_ID filter exists in filter_body and append the given dataflow_id to it (only if it's not already present).

Parameters:

Name Type Description Default
filter_body dict

e.g. {"filters":[{"type":"DATA_FLOW_ID","values":[116]}]}

 None
dataflow_id int

ID to add under the DATA_FLOW_ID filter

 None
Source code in src/crew_dcs/routes/dataflow/jupyter.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def generate_search_dataflows_to_jupyter_workspaces_body(
    filter_body: dict | None = None, dataflow_id: int | None = None
) -> dict:
    """
    Ensure the DATA_FLOW_ID filter exists in filter_body and append the given dataflow_id to it
    (only if it's not already present).

    Args:
        filter_body (dict): e.g. {"filters":[{"type":"DATA_FLOW_ID","values":[116]}]}
        dataflow_id (int): ID to add under the DATA_FLOW_ID filter
    """

    filter_body = filter_body or {}

    if not dataflow_id:
        return filter_body

    # 1. Make sure there's a filters list
    filters = filter_body.setdefault("filters", [])

    # 2. Try to find an existing DATA_FLOW_ID filter
    for f in filters:
        if f.get("type") == "DATA_FLOW_ID":
            # append if it's new
            if dataflow_id not in f["values"]:
                f["values"].append(dataflow_id)
            break
    else:
        # 3. If we never broke out, no filter existed—create it
        filters.append({"type": "DATA_FLOW_ID", "values": [dataflow_id]})

    return filter_body