Dataflows Gen1

create_dataflow_gen1(workspace, display_name, item_definition, *, description=None, folder=None, df=True)

Creates a new dataflow in the specified workspace.

Parameters:

Name	Type	Description	Default
workspace	str	The workspace name or ID.	required
display_name	str	The display name of the dataflow.	required
description	str	A description for the dataflow.	None
folder	str	The folder to create the dataflow in.	None
item_definition	Dict[str, Any]	The definition of the dataflow.	required
df	Optional[bool]	If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.	True

Returns:

Type	Description
Union[DataFrame, Dict[str, Any], None]	The created dataflow details.

Examples:

create_dataflow_gen1(
    workspace='MyProjectWorkspace',
    display_name='SalesDataflowGen1',
    item_definition={...},
    description='This is a sales dataflow',
    folder='SalesDataflowsFolder'
)

Source code in src/pyfabricops/items/dataflows_gen1.py

@df
def create_dataflow_gen1(
    workspace: str,
    display_name: str,
    item_definition: Dict[str, Any],
    *,
    description: Optional[str] = None,
    folder: Optional[str] = None,
    df: Optional[bool] = True,
) -> Union[DataFrame, Dict[str, Any], None]:
    """
    Creates a new dataflow in the specified workspace.

    Args:
        workspace (str): The workspace name or ID.
        display_name (str): The display name of the dataflow.
        description (str, optional): A description for the dataflow.
        folder (str, optional): The folder to create the dataflow in.
        item_definition (Dict[str, Any]): The definition of the dataflow.
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a list of dictionaries.

    Returns:
        (Union[DataFrame, Dict[str, Any], None]): The created dataflow details.

    Examples:
        ```python
        create_dataflow_gen1(
            workspace='MyProjectWorkspace',
            display_name='SalesDataflowGen1',
            item_definition={...},
            description='This is a sales dataflow',
            folder='SalesDataflowsFolder'
        )
        ```
    """
    workspace_id = resolve_workspace(workspace)

    payload = {
        'displayName': display_name,
        'definition': item_definition,
    }

    if folder:
        folder_id = resolve_folder(workspace_id, folder)
        if folder_id:
            payload['folderId'] = folder_id

    if description:
        payload['description'] = description

    return api_request(
        endpoint='/groups/' + workspace_id + '/dataflows',
        method='post',
        payload=payload,
        audience='powerbi',
    )

delete_dataflow_gen1(workspace, dataflow)

Deletes a dataflow from the specified workspace.

Parameters:

Name	Type	Description	Default
workspace	str	The name or ID of the workspace.	required
dataflow	str	The name or ID of the dataflow to delete.	required

Returns:

Name	Type	Description
None	None	If the dataflow is successfully deleted.

Examples:

delete_dataflow_gen1('MyProjectWorkspace', 'SalesDataflowGen1')
delete_dataflow_gen1('123e4567-e89b-12d3-a456-426614174000', 'SalesDataflowGen1')

Source code in src/pyfabricops/items/dataflows_gen1.py

def delete_dataflow_gen1(workspace: str, dataflow: str) -> None:
    """
    Deletes a dataflow from the specified workspace.

    Args:
        workspace (str): The name or ID of the workspace.
        dataflow (str): The name or ID of the dataflow to delete.

    Returns:
        None: If the dataflow is successfully deleted.

    Examples:
        ```python
        delete_dataflow_gen1('MyProjectWorkspace', 'SalesDataflowGen1')
        delete_dataflow_gen1('123e4567-e89b-12d3-a456-426614174000', 'SalesDataflowGen1')
        ```
    """
    workspace_id = resolve_workspace(workspace)

    dataflow_id = resolve_dataflow_gen1(workspace_id, dataflow)

    return api_request(
        endpoint='/groups/' + workspace_id + '/dataflows/' + dataflow_id,
        method='delete',
        audience='powerbi',
    )

get_dataflow_gen1(workspace, dataflow, *, df=True)

Get a Power BI dataflow.

Parameters:

Name	Type	Description	Default
workspace	str	The workspace name or ID.	required
dataflow	str	The name of the dataflow.	required
df	Optional[bool]	If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.	True

Returns:

Type	Description
Union[DataFrame, Dict[str, Any], None]	The dataflow.

Examples:

get_dataflow_gen1('MyProjectWorkspace', 'SalesDataflowGen1')
get_dataflow_gen1('123e4567-e89b-12d3-a456-426614174000', 'SalesDataflowGen1')

Source code in src/pyfabricops/items/dataflows_gen1.py

@df
def get_dataflow_gen1(
    workspace: str,
    dataflow: str,
    *,
    df: Optional[bool] = True,
) -> Union[DataFrame, Dict[str, Any], None]:
    """
    Get a Power BI dataflow.

    Args:
        workspace (str): The workspace name or ID.
        dataflow (str): The name of the dataflow.
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a list of dictionaries.

    Returns:
        (Union[DataFrame, Dict[str, Any], None]): The dataflow.

    Examples:
        ```python
        get_dataflow_gen1('MyProjectWorkspace', 'SalesDataflowGen1')
        get_dataflow_gen1('123e4567-e89b-12d3-a456-426614174000', 'SalesDataflowGen1')
        ```
    """
    workspace_id = resolve_workspace(workspace)

    return api_request(
        endpoint='/groups/'
        + workspace_id
        + '/dataflows/'
        + resolve_dataflow_gen1(workspace_id, dataflow),
        method='get',
        audience='powerbi',
    )

get_dataflow_gen1_definition(workspace, dataflow)

Get the definition of a Power BI dataflow Gen1.

Parameters:

Name	Type	Description	Default
workspace	str	The workspace name or ID.	required
dataflow	str	The dataflow name or ID.	required

Returns:

Name	Type	Description
dict	dict | None	The dataflow definition.

Examples:

get_dataflow_gen1_definition('MyProjectWorkspace', 'SalesDataflowGen1')
get_dataflow_gen1_definition('123e4567-e89b-12d3-a456-426614174000', 'SalesDataflowGen1')

Source code in src/pyfabricops/items/dataflows_gen1.py

def get_dataflow_gen1_definition(workspace: str, dataflow: str) -> dict | None:
    """
    Get the definition of a Power BI dataflow Gen1.

    Args:
        workspace (str): The workspace name or ID.
        dataflow (str): The dataflow name or ID.

    Returns:
        dict: The dataflow definition.

    Examples:
        ```python
        get_dataflow_gen1_definition('MyProjectWorkspace', 'SalesDataflowGen1')
        get_dataflow_gen1_definition('123e4567-e89b-12d3-a456-426614174000', 'SalesDataflowGen1')
        ```
    """
    workspace_id = resolve_workspace(workspace)

    dataflow_id = resolve_dataflow_gen1(workspace_id, dataflow)

    return api_request(
        endpoint='/groups/'
        + workspace_id
        + '/dataflows/'
        + dataflow_id
        + '/getDefinition',
        method='post',
        support_lro=True,
        audience='powerbi',
    )

get_dataflow_gen1_id(workspace, dataflow_name)

Retrieves the ID of a Gen1 dataflow by its name.

Parameters:

Name	Type	Description	Default
dataflow_name	str	The name of the dataflow.	required

Returns:

Type	Description
Union[str, None]	str | None: The ID of the dataflow if found, otherwise None.

Source code in src/pyfabricops/items/dataflows_gen1.py

def get_dataflow_gen1_id(
    workspace: str, dataflow_name: str
) -> Union[str, None]:
    """
    Retrieves the ID of a Gen1 dataflow by its name.

    Args:
        dataflow_name (str): The name of the dataflow.

    Returns:
        str | None: The ID of the dataflow if found, otherwise None.
    """
    dataflows = list_dataflows_gen1(
        workspace=resolve_workspace(workspace),
        df=False,
    )
    for _dataflow in dataflows:
        if _dataflow['displayName'] == dataflow_name:
            return _dataflow['id']
    logger.warning(
        f"Dataflow '{dataflow_name}' not found in workspace '{workspace}'."
    )
    return None

get_dataflow_gen1_transactions(workspace, dataflow, *, df=True)

Get transactions for a dataflow in a workspace.

Parameters:

Name	Type	Description	Default
workspace	str	The workspace name or ID.	required
dataflow	str	The dataflow name or ID.	required
df	Optional[bool]	If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.	True

Returns:

Type	Description
Union[DataFrame, List[Dict[str, Any]], None]	Union[DataFrame, List[Dict[str, Any]], None]: The dataflow transactions or None if not found.

Examples:

get_dataflow_gen1_transactions('MyProjectWorkspace', 'SalesDataflowGen1')
get_dataflow_gen1_transactions('123e4567-e89b-12d3-a456-426614174000', 'SalesDataflowGen1')

Source code in src/pyfabricops/items/dataflows_gen1.py

@df
def get_dataflow_gen1_transactions(
    workspace: str,
    dataflow: str,
    *,
    df: Optional[bool] = True,
) -> Union[DataFrame, List[Dict[str, Any]], None]:
    """
    Get transactions for a dataflow in a workspace.

    Args:
        workspace (str): The workspace name or ID.
        dataflow (str): The dataflow name or ID.
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a list of dictionaries.

    Returns:
        Union[DataFrame, List[Dict[str, Any]], None]: The dataflow transactions or None if not found.

    Examples:
        ```python
        get_dataflow_gen1_transactions('MyProjectWorkspace', 'SalesDataflowGen1')
        get_dataflow_gen1_transactions('123e4567-e89b-12d3-a456-426614174000', 'SalesDataflowGen1')
        ```
    """
    workspace_id = resolve_workspace(workspace)

    dataflow_id = resolve_dataflow_gen1(workspace_id, dataflow)

    return api_request(
        endpoint='/groups/'
        + workspace_id
        + '/dataflows/'
        + dataflow_id
        + '/transactions',
        method='post',
        audience='powerbi',
    )

get_dataflows_gen1_datasources(workspace, dataflow, *, df=True)

Get the data sources for a dataflow in a workspace.

Parameters:

Name	Type	Description	Default
workspace	str	The workspace name or ID.	required
dataflow	str	The dataflow name or ID.	required
df	Optional[bool]	If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.	True

Returns:

Type	Description
Union[DataFrame, List[Dict[str, Any]], None]	Union[DataFrame, List[Dict[str, Any]], None]: The dataflow datasources or None if not found.

Examples:

get_dataflows_gen1_datasources('MyProjectWorkspace', 'SalesDataflowGen1')
get_dataflows_gen1_datasources('123e4567-e89b-12d3-a456-426614174000', 'SalesDataflowGen1')

Source code in src/pyfabricops/items/dataflows_gen1.py

@df
def get_dataflows_gen1_datasources(
    workspace: str,
    dataflow: str,
    *,
    df: Optional[bool] = True,
) -> Union[DataFrame, List[Dict[str, Any]], None]:
    """
    Get the data sources for a dataflow in a workspace.

    Args:
        workspace (str): The workspace name or ID.
        dataflow (str): The dataflow name or ID.
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a list of dictionaries.

    Returns:
        Union[DataFrame, List[Dict[str, Any]], None]: The dataflow datasources or None if not found.

    Examples:
        ```python
        get_dataflows_gen1_datasources('MyProjectWorkspace', 'SalesDataflowGen1')
        get_dataflows_gen1_datasources('123e4567-e89b-12d3-a456-426614174000', 'SalesDataflowGen1')
        ```
    """
    workspace_id = resolve_workspace(workspace)

    dataflow_id = resolve_dataflow_gen1(workspace_id, dataflow)

    return api_request(
        endpoint='/groups/'
        + workspace_id
        + '/dataflows/'
        + dataflow_id
        + '/datasources',
        method='post',
        audience='powerbi',
    )

list_dataflows_gen1(workspace, *, df=True)

Returns a list of Gen1 dataflows in a specified workspace.

Parameters:

Name	Type	Description	Default
workspace	str	The name or ID of the workspace.	required
df	Optional[bool]	If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.	True

Returns:

Type	Description
Union[DataFrame, List[Dict[str, Any]], None]	A list of Gen1 dataflows or a DataFrame if df is True.

Source code in src/pyfabricops/items/dataflows_gen1.py

@df
def list_dataflows_gen1(
    workspace: str,
    *,
    df: Optional[bool] = True,
) -> Union[DataFrame, List[Dict[str, Any]], None]:
    """
    Returns a list of Gen1 dataflows in a specified workspace.

    Args:
        workspace (str): The name or ID of the workspace.
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a list of dictionaries.

    Returns:
        (Union[DataFrame, List[Dict[str, Any]], None]): A list of Gen1 dataflows or a DataFrame if df is True.
    """
    return api_request(
        endpoint='/groups/' + resolve_workspace(workspace) + '/dataflows',
        audience='powerbi',
        method='get',
        support_pagination=True,
    )

refresh_dataflow_gen1(workspace, dataflow, *, process_type='default', notify_option='NoNotification')

Refresh a dataflow in the specified workspace.

Parameters:

Name	Type	Description	Default
workspace	str	The workspace name or ID.	required
dataflow	str	The name or ID of the dataflow to refresh.	required
process_type	str	The process type to use for the refresh. Defaults to 'default'.	'default'
notify_option	Literal['MailOnFailure', 'NoNotification']	The notification option to use for the refresh. Defaults to 'NoNotification'.	'NoNotification'

Returns:

Name	Type	Description
None	None	If the refresh was successful.

Examples:

refresh_dataflow_gen1('MyProjectWorkspace', 'SalesDataflow')
refresh_dataflow_gen1('123e4567-e89b-12d3-a456-426614174000', 'SalesDataflow')

Source code in src/pyfabricops/items/dataflows_gen1.py

def refresh_dataflow_gen1(
    workspace: str,
    dataflow: str,
    *,
    process_type: Optional[str] = 'default',
    notify_option: Literal[
        'MailOnFailure', 'NoNotification'
    ] = 'NoNotification',
) -> None:
    """
    Refresh a dataflow in the specified workspace.

    Args:
        workspace (str): The workspace name or ID.
        dataflow (str): The name or ID of the dataflow to refresh.
        process_type (str, optional): The process type to use for the refresh. Defaults to 'default'.
        notify_option (Literal['MailOnFailure', 'NoNotification'], optional): The notification option to use for the refresh. Defaults to 'NoNotification'.

    Returns:
        None: If the refresh was successful.

    Examples:
        ```python
        refresh_dataflow_gen1('MyProjectWorkspace', 'SalesDataflow')
        refresh_dataflow_gen1('123e4567-e89b-12d3-a456-426614174000', 'SalesDataflow')
        ```
    """
    workspace_id = resolve_workspace(workspace)

    dataflow_id = resolve_dataflow_gen1(workspace_id, dataflow)

    payload = {'notifyOption': notify_option}

    params = {'processType': process_type}

    return api_request(
        endpoint='/groups/'
        + workspace_id
        + '/dataflows/'
        + dataflow_id
        + '/refreshes',
        method='post',
        payload=payload,
        params=params,
        audience='powerbi',
    )

resolve_dataflow_gen1(workspace, dataflow)

Resolves a dataflow name to its ID.

Parameters:

Name	Type	Description	Default
workspace	str	The name or ID of the workspace.	required
dataflow	str	The name or ID of the dataflow.	required

Returns:

Type	Description
Union[str, None]	str | None: The ID of the dataflow if found, otherwise None.

Source code in src/pyfabricops/items/dataflows_gen1.py

def resolve_dataflow_gen1(workspace: str, dataflow: str) -> Union[str, None]:
    """
    Resolves a dataflow name to its ID.

    Args:
        workspace (str): The name or ID of the workspace.
        dataflow (str): The name or ID of the dataflow.

    Returns:
        str | None: The ID of the dataflow if found, otherwise None.
    """
    if is_valid_uuid(dataflow):
        return dataflow
    else:
        return get_dataflow_gen1_id(workspace, dataflow)

takeover_dataflow_gen1(workspace, dataflow)

Take over a dataflow in a workspace

Parameters:

Name	Type	Description	Default
workspace	str	The workspace name or ID.	required
dataflow	str	The dataflow name or ID.	required

Examples:

takeover_dataflow_gen1('MyProjectWorkspace', 'SalesDataflowGen1')
takeover_dataflow_gen1('123e4567-e89b-12d3-a456-426614174000', 'SalesDataflowGen1')

Source code in src/pyfabricops/items/dataflows_gen1.py

def takeover_dataflow_gen1(workspace: str, dataflow: str) -> Union[bool, None]:
    """
    Take over a dataflow in a workspace

    Args:
        workspace (str): The workspace name or ID.
        dataflow (str): The dataflow name or ID.

    Examples:
        ```python
        takeover_dataflow_gen1('MyProjectWorkspace', 'SalesDataflowGen1')
        takeover_dataflow_gen1('123e4567-e89b-12d3-a456-426614174000', 'SalesDataflowGen1')
        ```
    """
    workspace_id = resolve_workspace(workspace)

    dataflow_id = resolve_dataflow_gen1(workspace_id, dataflow)

    return api_request(
        endpoint='/groups/'
        + workspace_id
        + '/dataflows/'
        + dataflow_id
        + '/Default.Takeover',
        method='post',
        support_lro=True,
    )

update_dataflow_gen1(workspace, dataflow, *, display_name=None, description=None, df=True)

Updates the properties of the specified dataflow.

Parameters:

Name	Type	Description	Default
workspace	str	The workspace name or ID.	required
dataflow	str	The name or ID of the dataflow to update.	required
display_name	str	The new display name for the dataflow.	None
description	str	The new description for the dataflow.	None
df	bool	Keyword-only. If True, returns a DataFrame with flattened keys. Defaults to False.	True

Returns:

Type	Description
Union[DataFrame, Dict[str, Any], None]	The updated dataflow details if successful, otherwise None.

Examples:

update_dataflow_gen1('MyProjectWorkspace', 'SalesDataflowGen1', display_name='UpdatedSalesDataflowGen1')
update_dataflow_gen1('MyProjectWorkspace', '123e4567-e89b-12d3-a456-426614174000', description='Updated description')

Source code in src/pyfabricops/items/dataflows_gen1.py

@df
def update_dataflow_gen1(
    workspace: str,
    dataflow: str,
    *,
    display_name: Optional[str] = None,
    description: Optional[str] = None,
    df: Optional[bool] = True,
) -> Union[DataFrame, Dict[str, Any], None]:
    """
    Updates the properties of the specified dataflow.

    Args:
        workspace (str): The workspace name or ID.
        dataflow (str): The name or ID of the dataflow to update.
        display_name (str, optional): The new display name for the dataflow.
        description (str, optional): The new description for the dataflow.
        df (bool, optional): Keyword-only. If True, returns a DataFrame with flattened keys. Defaults to False.

    Returns:
        (Union[DataFrame, Dict[str, Any], None]): The updated dataflow details if successful, otherwise None.

    Examples:
        ```python
        update_dataflow_gen1('MyProjectWorkspace', 'SalesDataflowGen1', display_name='UpdatedSalesDataflowGen1')
        update_dataflow_gen1('MyProjectWorkspace', '123e4567-e89b-12d3-a456-426614174000', description='Updated description')
        ```
    """
    workspace_id = resolve_workspace(workspace)

    dataflow_id = resolve_dataflow_gen1(workspace_id, dataflow)

    payload = {}

    if display_name:
        payload['displayName'] = display_name

    if description:
        payload['description'] = description

    return api_request(
        endpoint='/groups/' + workspace_id + '/dataflows/' + dataflow_id,
        method='patch',
        payload=payload,
        audience='powerbi',
    )

update_dataflow_gen1_definition(workspace, dataflow, item_definition)

Updates the definition of an existing dataflow in the specified workspace. If the dataflow does not exist, it returns None.

Parameters:

Name	Type	Description	Default
workspace	str	The workspace name or ID.	required
dataflow	str	The name or ID of the dataflow to update.	required
item_definition	Dict[str, Any]	The item_definition of the dataflow.	required
df	Optional[bool]	If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.	required

Returns:

Type	Description
Union[DataFrame, Dict[str, Any], None]	The updated dataflow details if successful, otherwise None.

Examples:

update_dataflow_gen1_definition(
    workspace='MyProjectWorkspace',
    dataflow='SalesDataflowGen1',
    item_definition={...} # The definition of the dataflow
)

Source code in src/pyfabricops/items/dataflows_gen1.py

@df
def update_dataflow_gen1_definition(
    workspace: str, dataflow: str, item_definition: Dict[str, Any]
) -> Union[DataFrame, Dict[str, Any], None]:
    """
    Updates the definition of an existing dataflow in the specified workspace.
    If the dataflow does not exist, it returns None.

    Args:
        workspace (str): The workspace name or ID.
        dataflow (str): The name or ID of the dataflow to update.
        item_definition (Dict[str, Any]): The item_definition of the dataflow.
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a list of dictionaries.

    Returns:
        (Union[DataFrame, Dict[str, Any], None]): The updated dataflow details if successful, otherwise None.

    Examples:
        ```python
        update_dataflow_gen1_definition(
            workspace='MyProjectWorkspace',
            dataflow='SalesDataflowGen1',
            item_definition={...} # The definition of the dataflow
        )
        ```
    """
    workspace_id = resolve_workspace(workspace)

    dataflow_id = resolve_dataflow_gen1(workspace_id, dataflow)

    params = {'updateMetadata': True}
    payload = {'definition': item_definition}

    return api_request(
        endpoint='/groups/' + workspace_id + '/dataflows/' + dataflow_id,
        method='patch',
        payload=payload,
        params=params,
        audience='powerbi',
    )