superwise_api.entities.policy

class PolicyApi(superwise_api.entities.base.BaseApi):
def create( self, name: str, data_config: Union[superwise_api.models.policy.policy.DataConfigStatistics, superwise_api.models.policy.policy.DataConfigDistributionCompare], cron_expression: str, threshold_settings: Union[superwise_api.models.policy.policy.StaticThresholdSettings, superwise_api.models.policy.policy.MovingAverageThresholdSettings], alert_on_status: superwise_api.models.policy.policy.AlertOnStatusDirection, alert_on_policy_level: bool, dataset_id: str, destination_ids: list[str], initialize_with_historic_data: bool = False, dataset_b_id: str = None, **kwargs) -> superwise_api.models.policy.policy.Policy:

Create a new policy.

Arguments:
  • name: The name of the policy.
  • data_config: The data configuration for the policy.
  • cron_expression: The cron expression that defines the schedule of the policy.
  • threshold_settings: The threshold settings for the policy.
  • alert_on_status: The direction of the alert.
  • alert_on_policy_level: Whether to alert on policy level.
  • dataset_id: The dataset id.
  • destination_ids: The destination ids.
  • initialize_with_historic_data: If true, the policy will be initialized with historic data
  • dataset_b_id: The second dataset this policy is monitoring.
  • **kwargs: Arbitrary keyword arguments.
Returns:

The created policy.

def get_by_id( self, policy_id: str, **kwargs) -> superwise_api.models.policy.policy.Policy:

Gets a policy by id.

Arguments:
  • policy_id (str): The id of the policy.
  • **kwargs: Arbitrary keyword arguments.
Returns:

Policy: The retrieved policy.

def get( self, name: Optional[str] = None, status: Optional[str] = None, created_by: Optional[str] = None, dataset_id: Optional[str] = None, page: Optional[int] = None, size: Optional[int] = None, **kwargs) -> superwise_api.client.models.page.Page:

Gets policies. Filter if any of the parameters are provided.

Arguments:
  • name (str, optional): The name of the policy.
  • status (str, optional): The status of the policy.
  • created_by (str, optional): The creator of the policy.
  • dataset_id (str, optional): The id of the dataset.
  • page (int, optional): The page number.
  • size (int, optional): The size of the page.
  • **kwargs: Arbitrary keyword arguments.
Returns:

Page: A page of policies.

def update( self, policy_id: str, *, name: Optional[str] = None, data_config: Optional[superwise_api.models.policy.policy.DataConfigBase] = None, cron_expression: Optional[str] = None, destination_ids: Optional[list[str]] = None, alert_on_status: Optional[superwise_api.models.policy.policy.AlertOnStatusDirection] = None, alert_on_policy_level: Optional[bool] = None, dataset_id: Optional[str] = None, threshold_settings: Union[superwise_api.models.policy.policy.StaticThresholdSettings, superwise_api.models.policy.policy.MovingAverageThresholdSettings, NoneType] = None, initialize_with_historic_data: Optional[bool] = None, **kwargs) -> superwise_api.models.policy.policy.Policy:

Update a policy.

Arguments:
  • policy_id: The id of the policy.
  • name: The name of the policy.
  • data_config: The data configuration for the policy.
  • cron_expression: The cron expression that defines the schedule of the policy.
  • threshold_settings: The threshold settings for the policy.
  • alert_on_status: The direction of the alert.
  • alert_on_policy_level: Whether to alert on policy level.
  • dataset_id: The dataset id.
  • destination_ids: The destination ids.
  • **kwargs: Arbitrary keyword arguments.
Returns:

The updated policy.

def delete(self, policy_id: str, **kwargs):

Deletes a policy.

Arguments:
  • policy_id (str): The id of the policy to delete.
  • **kwargs: Arbitrary keyword arguments.
@BaseApi.raise_exception
def trigger(self, policy_id: str, **kwargs):

Trigger a policy.

Arguments:
  • policy_id (str): The id of the policy to trigger.
  • **kwargs: Arbitrary keyword arguments.
def add_tags(self, policy_id: str, tag_ids: list[str], **kwargs):

Add tags to an policy.

Arguments:
  • policy_id (str): The id of the policy.
  • tag_ids list(str): List of tag ids to add the policy
Returns:

None

def remove_tags(self, policy_id: str, tag_ids: list[str], **kwargs):

Delete tags from an policy.

Arguments:
  • policy_id (str): The id of the policy.
  • tag_ids list(str): List of tag ids to remove from the policy
Returns:

None

def search( self, filters: list[typing.Any] | None = None, search: str | None = None, sort_by: str | None = None, sort_direction: Literal['asc', 'desc'] = 'desc', **kwargs) -> superwise_api.client.models.page.Page:

Searches for policies based on a prefix.

Arguments:
  • filters (list[Any]): Filter on db columns, list of tuples. e.g. [[["id", "eq", "5c05dc9f-f04a-4ce8-9d57-2ec63ee76aac"], "and", ["description", "ilike", "Construction"]], "or", ["name", "ilike", "active"]]
  • search (str): Free text search on searchable fields
  • sort_by (str): Field to sort by
  • sort_direction (Literal["asc", "desc"]): Sort direction (ascending or descending)
  • query_params can be passed as part of the kwargs for pagination
Returns:

Page: A page of policies.

Inherited Members
superwise_api.entities.base.BaseApi
BaseApi
api_client
wrap_api_calls
raise_exception