Skip to content

dataflow

dataflow

Dataflow route module - re-exports all dataflow functionality.

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

get_magic_etl_tile_documentation async 

get_magic_etl_tile_documentation(
    auth: DomoAuth,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve tile/function documentation for Magic ETL from the Domo API (v2 preferred).

Parameters:

Name Type Description Default
auth DomoAuth

Authentication object

 required
context RouteContext | None

RouteContext for request configuration

 None
**context_kwargs

Additional context parameters (session, debug_api, etc.)

 {}

Returns:

Type Description
ResponseGetData

ResponseGetData containing the Magic ETL tile documentation

Raises:

Type Description
MagicETLTileDocumentation_GET_Error

If the request fails
Source code in src/crew_dcs/routes/dataflow/magic_etl_tile_documentation.py 
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
57
58
59
60
61
62
63
64
65
66
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_magic_etl_tile_documentation(
    auth: DomoAuth,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve tile/function documentation for Magic ETL from the Domo API (v2 preferred).

    Args:
        auth: Authentication object
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData containing the Magic ETL tile documentation

    Raises:
        MagicETLTileDocumentation_GET_Error: If the request fails
    """
    url = f"https://{auth.domo_instance}.domo.com/api/dataprocessing/v2/expression-docs"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        context=context,
    )

    if not res.is_success:
        raise MagicETLTileDocumentation_GET_Error(res=res)

    return res

get_magic_etl_tile_documentation_v1 async 

get_magic_etl_tile_documentation_v1(
    auth: DomoAuth,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve tile/function documentation for Magic ETL from the Domo API (v1 fallback).

Parameters:

Name Type Description Default
auth DomoAuth

Authentication object

 required
context RouteContext | None

RouteContext for request configuration

 None
**context_kwargs

Additional context parameters (session, debug_api, etc.)

 {}

Returns:

Type Description
ResponseGetData

ResponseGetData containing the Magic ETL tile documentation

Raises:

Type Description
MagicETLTileDocumentation_GET_Error

If the request fails
Source code in src/crew_dcs/routes/dataflow/magic_etl_tile_documentation.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_magic_etl_tile_documentation_v1(
    auth: DomoAuth,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve tile/function documentation for Magic ETL from the Domo API (v1 fallback).

    Args:
        auth: Authentication object
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData containing the Magic ETL tile documentation

    Raises:
        MagicETLTileDocumentation_GET_Error: If the request fails
    """
    url = f"https://{auth.domo_instance}.domo.com/api/dataprocessing/v1/expression-docs"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        context=context,
    )

    if not res.is_success:
        raise MagicETLTileDocumentation_GET_Error(res=res)

    return res

Modules