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.
Inherited Members
superwise_api.entities.base.BaseApi
BaseApi
api_client
wrap_api_calls
raise_exception