pdp

pdp

PDP (Personalized Data Permissions) Route Functions

This module provides functions for managing Domo PDP policies including retrieval, creation, updating, and deletion operations. PDP policies control data access at the row level based on user, group, or virtual user assignments.

Submodules

exceptions: Exception classes for PDP operations core: Core retrieval and utility functions crud: Create, update, delete, and toggle operations

Exception Classes

PDP_GET_Error: Raised when PDP policy retrieval fails SearchPDPNotFoundError: Raised when PDP policy search returns no results PDP_CRUD_Error: Raised when PDP policy create/update/delete operations fail

Core Functions

get_pdp_policies: Retrieve all PDP policies for a dataset search_pdp_policies_by_name: Search for specific PDP policies by name generate_policy_parameter_simple: Utility function for creating policy parameters generate_policy_body: Utility function for creating policy request bodies

CRUD Functions

create_policy: Create a new PDP policy update_policy: Update an existing PDP policy delete_policy: Delete a PDP policy toggle_pdp: Enable or disable PDP for a dataset

PDP_CRUD_Error

PDP_CRUD_Error(
    operation: str,
    dataset_id: str | None = None,
    policy_id: str | None = None,
    res: ResponseGetData | None = None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

Raised when PDP policy create, update, or delete operations fail.

This exception is used for failures during policy creation, modification, or deletion operations.

Source code in src/crew_dcs/routes/pdp/exceptions.py

def __init__(
    self,
    operation: str,
    dataset_id: str | None = None,
    policy_id: str | None = None,
    res: rgd.ResponseGetData | None = None,
    message: str | None = None,
    **kwargs,
):
    if not message:
        if policy_id:
            message = f"PDP policy {operation} failed for policy {policy_id}"
        elif dataset_id:
            message = f"PDP policy {operation} failed for dataset {dataset_id}"
        else:
            message = f"PDP policy {operation} operation failed"

    super().__init__(
        message=message,
        entity_id=policy_id or dataset_id,
        res=res,
        **kwargs,
    )

PDP_GET_Error

PDP_GET_Error(
    dataset_id: str | None = None,
    res: ResponseGetData | None = None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

Raised when PDP policy retrieval operations fail.

This exception is used for failures during GET operations on PDP policies, including API errors and unexpected response formats.

Source code in src/crew_dcs/routes/pdp/exceptions.py

def __init__(
    self,
    dataset_id: str | None = None,
    res: rgd.ResponseGetData | None = None,
    message: str | None = None,
    **kwargs,
):
    if not message:
        if dataset_id:
            message = f"Failed to retrieve PDP policies for dataset {dataset_id}"
        else:
            message = "Failed to retrieve PDP policies"

    super().__init__(message=message, entity_id=dataset_id, res=res, **kwargs)

SearchPDPNotFoundError

SearchPDPNotFoundError(
    search_criteria: str,
    res: ResponseGetData | None = None,
    **kwargs
)

Bases: RouteError

Raised when PDP policy search operations return no results.

This exception is used when searching for specific PDP policies that don't exist or when search criteria match no policies.

Source code in src/crew_dcs/routes/pdp/exceptions.py

def __init__(
    self,
    search_criteria: str,
    res: rgd.ResponseGetData | None = None,
    **kwargs,
):
    message = f"No PDP policies found matching: {search_criteria}"
    super().__init__(
        message=message,
        res=res,
        **kwargs,
    )

create_policy async

create_policy(
    auth: DomoAuth,
    dataset_id: str,
    body: dict,
    override_same_name: bool = False,
    is_suppress_errors: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create a new PDP policy for a dataset.

Creates a new Personalized Data Permissions policy with the specified parameters and assignments. Can check for duplicate policy names before creating.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing instance and credentials	required
dataset_id	str	Unique identifier for the dataset	required
body	dict	Policy request body (from generate_policy_body)	required
override_same_name	bool	If True, allow creating policy with duplicate name	False
is_suppress_errors	bool	If True, return existing policy instead of error for duplicates	False
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	Optional RouteContext for request configuration	None
**context_kwargs		Additional context parameters (session, debug_api, etc.)	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing created policy information

Raises:

Type	Description
PDP_CRUD_Error	If policy creation fails or duplicate name exists

Example

params = [generate_policy_parameter_simple("Region", column_values_ls=["West"])] body = generate_policy_body( ... policy_name="West Region Access", ... dataset_id="abc123", ... parameters_ls=params ... ) response = await create_policy(auth, "abc123", body) policy_id = response.response.get("filterGroupId")

Source code in src/crew_dcs/routes/pdp/crud.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_policy(
    auth: DomoAuth,
    dataset_id: str,
    body: dict,
    override_same_name: bool = False,
    is_suppress_errors: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Create a new PDP policy for a dataset.

    Creates a new Personalized Data Permissions policy with the specified
    parameters and assignments. Can check for duplicate policy names before
    creating.

    Args:
        auth: Authentication object containing instance and credentials
        dataset_id: Unique identifier for the dataset
        body: Policy request body (from generate_policy_body)
        override_same_name: If True, allow creating policy with duplicate name
        is_suppress_errors: If True, return existing policy instead of error for duplicates
        return_raw: Return raw API response without processing
        context: Optional RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing created policy information

    Raises:
        PDP_CRUD_Error: If policy creation fails or duplicate name exists

    Example:
        >>> params = [generate_policy_parameter_simple("Region", column_values_ls=["West"])]
        >>> body = generate_policy_body(
        ...     policy_name="West Region Access",
        ...     dataset_id="abc123",
        ...     parameters_ls=params
        ... )
        >>> response = await create_policy(auth, "abc123", body)
        >>> policy_id = response.response.get("filterGroupId")
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/query/v1/data-control/{dataset_id}/filter-groups"

    if not override_same_name:
        existing_policies = await get_pdp_policies(
            auth=auth,
            dataset_id=dataset_id,
            context=context,
        )

        policy_exists = search_pdp_policies_by_name(
            search_name=body.get("name"),
            result_list=existing_policies.response,
            is_exact_match=True,
            is_suppress_errors=True,
        )

        if policy_exists:
            if not is_suppress_errors:
                raise PDP_CRUD_Error(
                    operation="create",
                    dataset_id=dataset_id,
                    res=existing_policies,
                    message='Policy name already exists. Avoid creating PDP policies with the same name. To override, set "override_same_name=True"',
                )

            return existing_policies

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

    if return_raw:
        return res

    if not res.is_success:
        raise PDP_CRUD_Error(
            operation="create",
            dataset_id=dataset_id,
            res=res,
            message=f"Failed to create policy - {res.response}",
        )

    return res

delete_policy async

delete_policy(
    auth: DomoAuth,
    dataset_id: str,
    policy_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Delete a PDP policy.

Permanently removes a Personalized Data Permissions policy from a dataset. This action cannot be undone.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing instance and credentials	required
dataset_id	str	Unique identifier for the dataset	required
policy_id	str	Unique identifier for the policy to delete	required
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	Optional RouteContext for request configuration	None
**context_kwargs		Additional context parameters (session, debug_api, etc.)	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object with confirmation message

Raises:

Type	Description
PDP_CRUD_Error	If policy deletion fails

Example

response = await delete_policy(auth, "abc123", "policy123") print(f"Policy deleted: {response.response}")

Source code in src/crew_dcs/routes/pdp/crud.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def delete_policy(
    auth: DomoAuth,
    dataset_id: str,
    policy_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Delete a PDP policy.

    Permanently removes a Personalized Data Permissions policy from a dataset.
    This action cannot be undone.

    Args:
        auth: Authentication object containing instance and credentials
        dataset_id: Unique identifier for the dataset
        policy_id: Unique identifier for the policy to delete
        return_raw: Return raw API response without processing
        context: Optional RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object with confirmation message

    Raises:
        PDP_CRUD_Error: If policy deletion fails

    Example:
        >>> response = await delete_policy(auth, "abc123", "policy123")
        >>> print(f"Policy deleted: {response.response}")
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/query/v1/data-control/{dataset_id}/filter-groups/{policy_id}"

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

    if return_raw:
        return res

    if not res.is_success:
        raise PDP_CRUD_Error(
            operation="delete",
            dataset_id=dataset_id,
            policy_id=policy_id,
            res=res,
            message=f"Failed to delete policy {policy_id} - {res.response}",
        )

    return res

generate_policy_body

generate_policy_body(
    policy_name: str,
    dataset_id: str,
    parameters_ls: list[dict],
    policy_id: str | None = None,
    user_ids: list[str] | None = None,
    group_ids: list[str] | None = None,
    virtual_user_ids: list[str] | None = None,
) -> dict

Generate a policy body for PDP policy creation or update.

Creates a complete request body for creating or updating a PDP policy, including filter parameters and user/group assignments.

Parameters:

Name	Type	Description	Default
policy_name	str	Name for the policy	required
dataset_id	str	Unique identifier for the dataset	required
parameters_ls	list[dict]	list of parameter dicts (from generate_policy_parameter_simple)	required
policy_id	str | None	Policy ID (only for updates, omit for new policies)	None
user_ids	list[str] | None	list of user IDs to assign the policy to	None
group_ids	list[str] | None	list of group IDs to assign the policy to	None
virtual_user_ids	list[str] | None	list of virtual user IDs to assign the policy to	None

Returns:

Type	Description
dict	Dictionary representing complete policy request body

Example

params = [generate_policy_parameter_simple("Region", column_values_ls=["West"])] body = generate_policy_body( ... policy_name="West Region Access", ... dataset_id="abc123", ... parameters_ls=params, ... user_ids=["12345"] ... )

Use body in create_policy or update_policy

Source code in src/crew_dcs/routes/pdp/core.py

def generate_policy_body(
    policy_name: str,
    dataset_id: str,
    parameters_ls: list[dict],
    policy_id: str | None = None,
    user_ids: list[str] | None = None,
    group_ids: list[str] | None = None,
    virtual_user_ids: list[str] | None = None,
) -> dict:
    """
    Generate a policy body for PDP policy creation or update.

    Creates a complete request body for creating or updating a PDP policy,
    including filter parameters and user/group assignments.

    Args:
        policy_name: Name for the policy
        dataset_id: Unique identifier for the dataset
        parameters_ls: list of parameter dicts (from generate_policy_parameter_simple)
        policy_id: Policy ID (only for updates, omit for new policies)
        user_ids: list of user IDs to assign the policy to
        group_ids: list of group IDs to assign the policy to
        virtual_user_ids: list of virtual user IDs to assign the policy to

    Returns:
        Dictionary representing complete policy request body

    Example:
        >>> params = [generate_policy_parameter_simple("Region", column_values_ls=["West"])]
        >>> body = generate_policy_body(
        ...     policy_name="West Region Access",
        ...     dataset_id="abc123",
        ...     parameters_ls=params,
        ...     user_ids=["12345"]
        ... )
        >>> # Use body in create_policy or update_policy
    """
    if not user_ids:
        user_ids = []

    if not group_ids:
        group_ids = []

    if not virtual_user_ids:
        virtual_user_ids = []

    if not isinstance(parameters_ls, list):
        parameters_ls = [parameters_ls]

    body = {
        "name": policy_name,
        "dataSourceId": dataset_id,
        "userIds": user_ids,
        "virtualUserIds": virtual_user_ids,
        "groupIds": group_ids,
        "dataSourcePermissions": False,
        "parameters": parameters_ls,
    }

    if policy_id:
        body.update({"filterGroupId": policy_id})

    return body

generate_policy_parameter_simple

generate_policy_parameter_simple(
    column_name: str,
    type: str = "COLUMN",
    column_values_ls: list[str] | None = None,
    operator: str = "EQUALS",
    ignore_case: bool = True,
) -> dict

Generate a simple policy parameter for PDP policy creation.

Creates a parameter dictionary that defines a filter condition for a PDP policy. Parameters specify which column values users can see.

Parameters:

Name	Type	Description	Default
column_name	str	Name of the column to filter on	required
type	str	Parameter type (default: "COLUMN")	'COLUMN'
column_values_ls	list[str] | None	list of column values to filter, or single value	None
operator	str	Comparison operator (default: "EQUALS")	'EQUALS'
ignore_case	bool	Whether to ignore case when comparing values (default: True)	True

Returns:

Type	Description
dict	Dictionary representing a policy parameter

Example

param = generate_policy_parameter_simple( ... column_name="Region", ... column_values_ls=["West", "East"] ... ) print(param) {'type': 'COLUMN', 'name': 'Region', 'values': ['West', 'East'], ...}

Source code in src/crew_dcs/routes/pdp/core.py

def generate_policy_parameter_simple(
    column_name: str,
    type: str = "COLUMN",
    column_values_ls: list[str] | None = None,
    operator: str = "EQUALS",
    ignore_case: bool = True,
) -> dict:
    """
    Generate a simple policy parameter for PDP policy creation.

    Creates a parameter dictionary that defines a filter condition for a PDP policy.
    Parameters specify which column values users can see.

    Args:
        column_name: Name of the column to filter on
        type: Parameter type (default: "COLUMN")
        column_values_ls: list of column values to filter, or single value
        operator: Comparison operator (default: "EQUALS")
        ignore_case: Whether to ignore case when comparing values (default: True)

    Returns:
        Dictionary representing a policy parameter

    Example:
        >>> param = generate_policy_parameter_simple(
        ...     column_name="Region",
        ...     column_values_ls=["West", "East"]
        ... )
        >>> print(param)
        {'type': 'COLUMN', 'name': 'Region', 'values': ['West', 'East'], ...}
    """
    if not isinstance(column_values_ls, list):
        column_values_ls = [column_values_ls] if column_values_ls is not None else []

    return {
        "type": type,
        "name": column_name,
        "values": column_values_ls,
        "operator": operator,
        "ignoreCase": ignore_case,
    }

get_pdp_policies async

get_pdp_policies(
    auth: DomoAuth,
    dataset_id: str,
    include_all_rows: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve all PDP policies for a specific dataset.

Fetches a list of all PDP (Personalized Data Permissions) policies associated with the specified dataset. Includes policy filters, associations, and open policy settings when include_all_rows is True.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing instance and credentials	required
dataset_id	str	Unique identifier for the dataset	required
include_all_rows	bool	Include policy associations, filters, and open policy (default: True)	True
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	Optional RouteContext for request configuration	None
**context_kwargs		Additional context parameters (session, debug_api, etc.)	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing list of PDP policies

Raises:

Type	Description
PDP_GET_Error	If PDP policy retrieval fails or API returns an error

Example

policies_response = await get_pdp_policies(auth, "abc123") for policy in policies_response.response: ... print(f"Policy: {policy['name']}, ID: {policy['filterGroupId']}")

Source code in src/crew_dcs/routes/pdp/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_pdp_policies(
    auth: DomoAuth,
    dataset_id: str,
    include_all_rows: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Retrieve all PDP policies for a specific dataset.

    Fetches a list of all PDP (Personalized Data Permissions) policies associated
    with the specified dataset. Includes policy filters, associations, and open
    policy settings when include_all_rows is True.

    Args:
        auth: Authentication object containing instance and credentials
        dataset_id: Unique identifier for the dataset
        include_all_rows: Include policy associations, filters, and open policy (default: True)
        return_raw: Return raw API response without processing
        context: Optional RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing list of PDP policies

    Raises:
        PDP_GET_Error: If PDP policy retrieval fails or API returns an error

    Example:
        >>> policies_response = await get_pdp_policies(auth, "abc123")
        >>> for policy in policies_response.response:
        ...     print(f"Policy: {policy['name']}, ID: {policy['filterGroupId']}")
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"http://{auth.domo_instance}.domo.com/api/query/v1/data-control/{dataset_id}/filter-groups/"

    if include_all_rows:
        url += "?options=load_associations,load_filters,include_open_policy"

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

    if return_raw:
        return res

    if not res.is_success or (
        isinstance(res.response, list) and len(res.response) == 0
    ):
        raise PDP_GET_Error(
            dataset_id=dataset_id,
            res=res,
            message=f"Failed to retrieve PDP policies for dataset {dataset_id}",
        )

    return res

search_pdp_policies_by_name

search_pdp_policies_by_name(
    search_name: str,
    result_list: list[dict],
    is_exact_match: bool = True,
    is_suppress_errors: bool = False,
) -> dict | list[dict | bool]

Search for PDP policies by name within a list of policies.

Searches through a list of PDP policies to find those matching the specified name. Can perform exact or partial matching.

Parameters:

Name	Type	Description	Default
search_name	str	Name or partial name to search for	required
result_list	list[dict]	list of policy dictionaries from get_pdp_policies response	required
is_exact_match	bool	If True, search for exact name match; if False, partial match	True
is_suppress_errors	bool	If True, return False instead of raising error when not found	False

Returns:

Type	Description
dict | list[dict | bool]	Single policy dict (exact match), list of policy dicts (partial match),
dict | list[dict | bool]	or False if no matches and is_suppress_errors is True

Raises:

Type	Description
SearchPDPNotFoundError	If no policies match the search criteria (unless is_suppress_errors is True)

Example

policies = await get_pdp_policies(auth, "abc123") policy = search_pdp_policies_by_name("Sales Policy", policies.response) print(f"Found policy: {policy['filterGroupId']}")

Source code in src/crew_dcs/routes/pdp/core.py

def search_pdp_policies_by_name(
    search_name: str,
    result_list: list[dict],
    is_exact_match: bool = True,
    is_suppress_errors: bool = False,
) -> dict | list[dict | bool]:
    """
    Search for PDP policies by name within a list of policies.

    Searches through a list of PDP policies to find those matching the specified
    name. Can perform exact or partial matching.

    Args:
        search_name: Name or partial name to search for
        result_list: list of policy dictionaries from get_pdp_policies response
        is_exact_match: If True, search for exact name match; if False, partial match
        is_suppress_errors: If True, return False instead of raising error when not found

    Returns:
        Single policy dict (exact match), list of policy dicts (partial match),
        or False if no matches and is_suppress_errors is True

    Raises:
        SearchPDPNotFoundError: If no policies match the search criteria (unless is_suppress_errors is True)

    Example:
        >>> policies = await get_pdp_policies(auth, "abc123")
        >>> policy = search_pdp_policies_by_name("Sales Policy", policies.response)
        >>> print(f"Found policy: {policy['filterGroupId']}")
    """
    if is_exact_match:
        policy_search = next(
            (policy for policy in result_list if policy["name"] == search_name), None
        )
    else:
        policy_search = [
            policy
            for policy in result_list
            if search_name.lower() in policy["name"].lower()
        ]

    if not policy_search and not is_suppress_errors:
        raise SearchPDPNotFoundError(
            search_criteria=f"name: {search_name}",
        )

    return policy_search or False

toggle_pdp async

toggle_pdp(
    auth: DomoAuth,
    dataset_id: str,
    is_enable: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Enable or disable PDP for a dataset.

Toggles Personalized Data Permissions on or off for the specified dataset. When disabled, all users can see all data in the dataset.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing instance and credentials	required
dataset_id	str	Unique identifier for the dataset	required
is_enable	bool	If True, enable PDP; if False, disable PDP (default: True)	True
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	Optional RouteContext for request configuration	None
**context_kwargs		Additional context parameters (session, debug_api, etc.)	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object with confirmation message

Raises:

Type	Description
PDP_CRUD_Error	If toggle operation fails

Example

Enable PDP for a dataset

response = await toggle_pdp(auth, "abc123", is_enable=True)

Disable PDP for a dataset

response = await toggle_pdp(auth, "abc123", is_enable=False)

Source code in src/crew_dcs/routes/pdp/crud.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def toggle_pdp(
    auth: DomoAuth,
    dataset_id: str,
    is_enable: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Enable or disable PDP for a dataset.

    Toggles Personalized Data Permissions on or off for the specified dataset.
    When disabled, all users can see all data in the dataset.

    Args:
        auth: Authentication object containing instance and credentials
        dataset_id: Unique identifier for the dataset
        is_enable: If True, enable PDP; if False, disable PDP (default: True)
        return_raw: Return raw API response without processing
        context: Optional RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object with confirmation message

    Raises:
        PDP_CRUD_Error: If toggle operation fails

    Example:
        >>> # Enable PDP for a dataset
        >>> response = await toggle_pdp(auth, "abc123", is_enable=True)
        >>> # Disable PDP for a dataset
        >>> response = await toggle_pdp(auth, "abc123", is_enable=False)
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = (
        f"https://{auth.domo_instance}.domo.com/api/query/v1/data-control/{dataset_id}"
    )

    body = {
        "enabled": is_enable,
        "external": False,  # not sure what this parameter does
    }

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

    if return_raw:
        return res

    if not res.is_success:
        action = "enable" if is_enable else "disable"
        raise PDP_CRUD_Error(
            operation=f"toggle ({action})",
            dataset_id=dataset_id,
            res=res,
            message=f"Failed to {action} PDP for dataset {dataset_id} - {res.response}",
        )

    return res

update_policy async

update_policy(
    auth: DomoAuth,
    dataset_id: str,
    policy_id: str,
    body: dict,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update an existing PDP policy.

Modifies an existing Personalized Data Permissions policy with new parameters, assignments, or name.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing instance and credentials	required
dataset_id	str	Unique identifier for the dataset	required
policy_id	str	Unique identifier for the policy to update	required
body	dict	Policy request body (from generate_policy_body)	required
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	Optional RouteContext for request configuration	None
**context_kwargs		Additional context parameters (session, debug_api, etc.)	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing updated policy information

Raises:

Type	Description
PDP_CRUD_Error	If policy update fails

Example

params = [generate_policy_parameter_simple("Region", column_values_ls=["West", "East"])] body = generate_policy_body( ... policy_name="Updated Policy Name", ... dataset_id="abc123", ... parameters_ls=params, ... policy_id="policy123" ... ) response = await update_policy(auth, "abc123", "policy123", body)

Source code in src/crew_dcs/routes/pdp/crud.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def update_policy(
    auth: DomoAuth,
    dataset_id: str,
    policy_id: str,
    body: dict,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Update an existing PDP policy.

    Modifies an existing Personalized Data Permissions policy with new
    parameters, assignments, or name.

    Args:
        auth: Authentication object containing instance and credentials
        dataset_id: Unique identifier for the dataset
        policy_id: Unique identifier for the policy to update
        body: Policy request body (from generate_policy_body)
        return_raw: Return raw API response without processing
        context: Optional RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing updated policy information

    Raises:
        PDP_CRUD_Error: If policy update fails

    Example:
        >>> params = [generate_policy_parameter_simple("Region", column_values_ls=["West", "East"])]
        >>> body = generate_policy_body(
        ...     policy_name="Updated Policy Name",
        ...     dataset_id="abc123",
        ...     parameters_ls=params,
        ...     policy_id="policy123"
        ... )
        >>> response = await update_policy(auth, "abc123", "policy123", body)
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/query/v1/data-control/{dataset_id}/filter-groups/{policy_id}"

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

    if return_raw:
        return res

    if not res.is_success:
        raise PDP_CRUD_Error(
            operation="update",
            dataset_id=dataset_id,
            policy_id=policy_id,
            res=res,
            message=f"Failed to update policy {policy_id} - {res.response}",
        )

    return res

Modules

• core
• crud
• exceptions