pdp

pdp

PDP (Personalized Data Permissions) system using the unified entities architecture.

This module models PDP policies as entities and parameters. Relationship associations are pending refactor to the new Relationships system.

DatasetPdpPolicies dataclass

DatasetPdpPolicies(
    parent: DomoEntity,
    policies: list[PDPPolicy] = list(),
    is_pdp_enabled: bool = False,
)

Bases: DomoSubEntity

Manager for all PDP policies associated with a dataset.

This class provides high-level operations for managing the collection of PDP policies that apply to a specific dataset.

Attributes:

Name	Type	Description
dataset_id		ID of the dataset these policies apply to
policies	list[PDPPolicy]	list of PDP policies for this dataset
is_pdp_enabled	bool	Whether PDP is enabled for this dataset

add_policy

add_policy(policy: PDPPolicy) -> None

Add a policy to this dataset.

Source code in src/crew_dcs/classes/DomoDataset/pdp.py

def add_policy(self, policy: PDPPolicy) -> None:
    """Add a policy to this dataset."""
    policy.dataset_id = self.dataset_id
    self.policies.append(policy)

get async

get(
    auth: DomoAuth | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Fetch and hydrate PDP policies for this dataset.

Returns the hydrated policy list by default, or raw route response when return_raw=True.

Source code in src/crew_dcs/classes/DomoDataset/pdp.py

async def get(
    self,
    auth: DomoAuth | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Fetch and hydrate PDP policies for this dataset.

    Returns the hydrated policy list by default, or raw route response when
    return_raw=True.
    """
    auth = auth or self.auth
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await dataset_routes.get_dataset_policies(
        auth=auth,
        dataset_id=self.dataset_id,
        context=context,
    )

    if return_raw:
        return res

    self.policies = []
    for policy_data in res.response:
        self.add_policy(PDPPolicy.from_api_dict(policy_data))

    return self.policies

get_active_policies

get_active_policies() -> list[PDPPolicy]

Get all currently active policies.

Source code in src/crew_dcs/classes/DomoDataset/pdp.py

def get_active_policies(self) -> list[PDPPolicy]:
    """Get all currently active policies."""
    return [p for p in self.policies if p.is_active()]

get_policy_by_id

get_policy_by_id(policy_id: str) -> PDPPolicy | None

Get a policy by its ID.

Source code in src/crew_dcs/classes/DomoDataset/pdp.py

def get_policy_by_id(self, policy_id: str) -> PDPPolicy | None:
    """Get a policy by its ID."""
    return next(
        (
            p
            for p in self.policies
            if p.id == policy_id or p.filter_group_id == policy_id
        ),
        None,
    )

get_policy_by_name

get_policy_by_name(
    name: str, exact_match: bool = True
) -> PDPPolicy | None

Get a policy by its name.

Source code in src/crew_dcs/classes/DomoDataset/pdp.py

def get_policy_by_name(
    self, name: str, exact_match: bool = True
) -> PDPPolicy | None:
    """Get a policy by its name."""
    if exact_match:
        return next((p for p in self.policies if p.name == name), None)
    return next((p for p in self.policies if name.lower() in p.name.lower()), None)

remove_policy

remove_policy(policy_id: str) -> None

Remove a policy from this dataset.

Source code in src/crew_dcs/classes/DomoDataset/pdp.py

def remove_policy(self, policy_id: str) -> None:
    """Remove a policy from this dataset."""
    self.policies = [
        p
        for p in self.policies
        if p.id != policy_id and p.filter_group_id != policy_id
    ]

PDPOperator

Bases: DomoEnumMixin, Enum

Operators available for PDP policy parameters.

PDPParameterType

Bases: DomoEnumMixin, Enum

Types of PDP policy parameters.

PDPPolicy dataclass

PDPPolicy(
    auth: DomoAuth,
    id: str,
    raw: dict,
    dataset_id: str | None = None,
    filter_group_id: str | None = None,
    parameters: list[PdpParameter] = list(),
    policy_status: PDPPolicyStatus | None = ACTIVE,
    is_enabled: bool = True,
    priority: int = 1,
    effective_date: datetime | None = None,
    expiration_date: datetime | None = None,
    created_by_user_id: str | None = None,
    last_modified_by_user_id: str | None = None,
)

Bases: DomoEntity

Represents a PDP (Personalized Data Permissions) policy.

A PDP policy is a rule that applies to a dataset, defining how data should be filtered based on parameters and which users/groups have access.

The policy uses the relationship system to manage user and group associations rather than storing direct lists, providing better tracking and management.

Attributes:

Name	Type	Description
dataset_id	str | None	ID of the dataset this policy applies to
filter_group_id	str | None	Domo's internal ID for this policy
parameters	list[PdpParameter]	list of filter parameters that define the data restrictions
policy_status	PDPPolicyStatus | None	Current status of the policy (active, inactive, etc.)
is_enabled	bool	Whether the policy is currently enforced
priority	int	Priority order when multiple policies apply
effective_date	datetime | None	When the policy becomes effective
expiration_date	datetime | None	When the policy expires (if applicable)
created_by_user_id	str | None	ID of the user who created this policy
last_modified_by_user_id	str | None	ID of the user who last modified this policy

add_parameter

add_parameter(parameter: PdpParameter) -> None

Add a filter parameter to this policy.

Source code in src/crew_dcs/classes/DomoDataset/pdp.py

def add_parameter(self, parameter: PdpParameter) -> None:
    """Add a filter parameter to this policy."""
    parameter.parent_id = self.id or self.filter_group_id
    self.parameters.append(parameter)

from_api_dict classmethod

from_api_dict(data: dict[str, Any]) -> PDPPolicy

Create policy from API response data.

Source code in src/crew_dcs/classes/DomoDataset/pdp.py

@classmethod
def from_api_dict(cls, data: dict[str, Any]) -> "PDPPolicy":
    """Create policy from API response data."""
    policy = cls(
        id=data.get("filterGroupId"),
        filter_group_id=data.get("filterGroupId"),
        name=data.get("name"),
        dataset_id=data.get("dataSourceId"),
        is_enabled=data.get("enabled", True),
        priority=data.get("priority", 1),
    )

    # Add parameters
    for param_data in data.get("parameters", []):
        parameter = PdpParameter.from_api_dict(param_data)
        policy.add_parameter(parameter)

    return policy

is_active

is_active() -> bool

Check if the policy is currently active and effective.

Source code in src/crew_dcs/classes/DomoDataset/pdp.py

def is_active(self) -> bool:
    """Check if the policy is currently active and effective."""
    if not self.is_enabled or self.policy_status != PDPPolicyStatus.ACTIVE:
        return False

    now = datetime.now()

    if self.effective_date and self.effective_date > now:
        return False

    return not (self.expiration_date and self.expiration_date < now)

remove_parameter

remove_parameter(parameter_id: str) -> None

Remove a parameter from this policy.

Source code in src/crew_dcs/classes/DomoDataset/pdp.py

def remove_parameter(self, parameter_id: str) -> None:
    """Remove a parameter from this policy."""
    self.parameters = [p for p in self.parameters if p.id != parameter_id]

to_api_dict

to_api_dict() -> dict[str, Any]

Convert policy to API format for Domo requests.

Source code in src/crew_dcs/classes/DomoDataset/pdp.py

def to_api_dict(self) -> dict[str, Any]:
    """Convert policy to API format for Domo requests."""
    return {
        "filterGroupId": self.filter_group_id,
        "name": self.name,
        "dataSourceId": self.dataset_id,
        "parameters": [param.to_api_dict() for param in self.parameters],
        "userIds": [],
        "groupIds": [],
        "virtualUserIds": [],  # Legacy field, typically empty
        "enabled": self.is_enabled,
        "priority": self.priority,
    }

PDPPolicyStatus

Bases: DomoEnumMixin, Enum

Status of PDP policies.

PdpParameter dataclass

PdpParameter(
    parent: DomoEntity,
    column_name: str | None = None,
    column_values: list[str] = list(),
    operator: PDPOperator | None = EQUALS,
    parameter_type: PDPParameterType | None = COLUMN,
    ignore_case: bool = True,
    is_required: bool = True,
    description: str | None = None,
    trusted_attribute_key: str | None = None,
)

Bases: DomoSubEntity

Represents a single parameter (filter) within a PDP policy.

Each parameter defines a specific filter condition that will be applied to the dataset when users access it through the policy.

Attributes:

Name	Type	Description
column_name	str | None	The dataset column this parameter filters on
column_values	list[str]	list of values to filter by
operator	PDPOperator | None	The comparison operator to use (equals, greater than, etc.)
parameter_type	PDPParameterType | None	Whether this is a column-based or dynamic filter
ignore_case	bool	Whether string comparisons should ignore case
is_required	bool	Whether this parameter must have a value
description	str | None	Human-readable description of what this parameter does
trusted_attribute_key	str | None	For dynamic parameters, the trusted attribute to use

from_api_dict classmethod

from_api_dict(data: dict[str, Any]) -> PdpParameter

Create parameter from API response data.

Source code in src/crew_dcs/classes/DomoDataset/pdp.py

@classmethod
def from_api_dict(cls, data: dict[str, Any]) -> "PdpParameter":
    """Create parameter from API response data."""
    return cls(
        column_name=data.get("columnName"),
        column_values=data.get("values", []),
        operator=PDPOperator.get(data.get("operator", "EQUALS")),
        parameter_type=PDPParameterType.get(data.get("type", "COLUMN")),
        ignore_case=data.get("ignoreCase", True),
        trusted_attribute_key=data.get("trustedAttributeKey"),
    )

to_api_dict

to_api_dict() -> dict[str, Any]

Convert parameter to API format for Domo requests.

Source code in src/crew_dcs/classes/DomoDataset/pdp.py

def to_api_dict(self) -> dict[str, Any]:
    """Convert parameter to API format for Domo requests."""
    return {
        "columnName": self.column_name,
        "values": self.column_values,
        "operator": (
            self.operator.value if self.operator else PDPOperator.EQUALS.value
        ),
        "type": (
            self.parameter_type.value
            if self.parameter_type
            else PDPParameterType.COLUMN.value
        ),
        "ignoreCase": self.ignore_case,
        "trustedAttributeKey": self.trusted_attribute_key,
    }