Manager

psengine.playbook_alerts.playbook_alert_mgr.PlaybookAlertMgr

PlaybookAlertMgr(rf_token: Optional[str] = None)

Manages requests for Recorded Future playbook alerts.

PARAMETER	DESCRIPTION
rf_token	Recorded Future API token. TYPE: Optional[str] DEFAULT: None

Source code in psengine/playbook_alerts/playbook_alert_mgr.py

def __init__(
    self,
    rf_token: Annotated[Optional[str], Doc('Recorded Future API token.')] = None,
):
    """Initialize the `PlaybookAlertMgr` object."""
    self.log = logging.getLogger(__name__)
    self.rf_client = RFClient(api_token=rf_token) if rf_token is not None else RFClient()

fetch

fetch(
    alert_id: str,
    category: Optional[PACategory] = None,
    panels: Optional[list[str]] = None,
    fetch_images: Optional[bool] = True,
) -> PLAYBOOK_ALERT_TYPE

Fetch an individual Playbook Alert.

PARAMETER	DESCRIPTION
alert_id	Alert ID to fetch. TYPE: str
category	Category to fetch. If not given, playbook-alert/common is used to determine it. TYPE: Optional[PACategory] DEFAULT: None
panels	Panels to fetch. The status panel is always fetched for ADT initialization. TYPE: Optional[list[str]] DEFAULT: None
fetch_images	Fetch images for Domain Abuse & Geopol alerts. TYPE: Optional[bool] DEFAULT: True

Endpoints

• playbook-alert/{category}
• playbook-alert/common/{alert_id}

RAISES	DESCRIPTION
ValidationError	If any parameter is of incorrect type.
PlaybookAlertFetchError	If an API-related error occurs.

RETURNS	DESCRIPTION
PLAYBOOK_ALERT_TYPE	One of the Playbook Alert ADTs returned from the API.

Source code in psengine/playbook_alerts/playbook_alert_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=PlaybookAlertFetchError)
def fetch(
    self,
    alert_id: Annotated[str, Doc('Alert ID to fetch.')],
    category: Annotated[
        Optional[PACategory],
        Doc(
            'Category to fetch. If not given, `playbook-alert/common` is used to determine it.'
        ),
    ] = None,
    panels: Annotated[
        Optional[list[str]],
        Doc('Panels to fetch. The `status` panel is always fetched for ADT initialization.'),
    ] = None,
    fetch_images: Annotated[
        Optional[bool], Doc('Fetch images for Domain Abuse & Geopol alerts.')
    ] = True,
) -> Annotated[
    PLAYBOOK_ALERT_TYPE,
    Doc('One of the Playbook Alert ADTs returned from the API.'),
]:
    """Fetch an individual Playbook Alert.

    Endpoints:
        - `playbook-alert/{category}`
        - `playbook-alert/common/{alert_id}`

    Raises:
        ValidationError: If any parameter is of incorrect type.
        PlaybookAlertFetchError: If an API-related error occurs.
    """
    if category is None:
        category = self._fetch_alert_category(alert_id)

    category = category.lower()

    data = {}
    if panels:
        # We must always fetch status panel for ADT initialization
        if STATUS_PANEL_NAME not in panels:
            panels.append(STATUS_PANEL_NAME)
        data = {'panels': panels}

    url = f'{CATEGORY_ENDPOINTS[category]}/{alert_id}'
    self.log.info(f'Fetching playbook alert: {alert_id}, category: {category}')

    response = self.rf_client.request('post', url=url, data=data)
    p_alert = self._playbook_alert_factory(category, response.json()['data'])

    if isinstance(p_alert, PBA_WITH_IMAGES_INST) and fetch_images:
        self.fetch_images(p_alert)

    return p_alert

fetch_bulk

fetch_bulk(
    alerts: Optional[list[tuple[str, PACategory]]] = None,
    panels: Optional[list[str]] = None,
    fetch_images: Optional[bool] = False,
    alerts_per_page: Optional[int] = Field(
        ge=1, le=10000, default=ALERTS_PER_PAGE
    ),
    max_results: Optional[int] = DEFAULT_LIMIT,
    order_by: Optional[str] = None,
    direction: Optional[str] = None,
    entity: Union[str, list, None] = None,
    statuses: Union[str, list, None] = None,
    priority: Union[str, list, None] = None,
    category: Union[
        PACategory, list[PACategory], None
    ] = None,
    assignee: Union[str, list, None] = None,
    created_from: Optional[str] = None,
    created_until: Optional[str] = None,
    updated_from: Optional[str] = None,
    updated_until: Optional[str] = None,
) -> list[PLAYBOOK_ALERT_TYPE]

Fetch multiple playbook alerts in bulk, by query filters or specified alert tuples.

PARAMETER	DESCRIPTION
alerts	List of (alert_id, category) tuples to fetch. Overrides search parameters. TYPE: Optional[list[tuple[str, PACategory]]] DEFAULT: None
panels	Panels to fetch for each alert. The status panel is always fetched. TYPE: Optional[list[str]] DEFAULT: None
fetch_images	Whether to fetch images for supported alert types. TYPE: Optional[bool] DEFAULT: False
alerts_per_page	Number of alerts per page (pagination). TYPE: Optional[int] DEFAULT: Field(ge=1, le=10000, default=ALERTS_PER_PAGE)
max_results	Maximum number of alerts to fetch. TYPE: Optional[int] DEFAULT: DEFAULT_LIMIT
order_by	Field to order alerts by, e.g. created or updated. TYPE: Optional[str] DEFAULT: None
direction	Sort direction: asc or desc. TYPE: Optional[str] DEFAULT: None
entity	Entity or list of entities to filter alerts by. TYPE: Union[str, list, None] DEFAULT: None
statuses	Status or list of statuses to filter alerts by, e.g. ['New', 'Closed']. TYPE: Union[str, list, None] DEFAULT: None
priority	Priority or list of priorities, e.g. ['High', 'Low']. TYPE: Union[str, list, None] DEFAULT: None
category	Category or list of categories to filter alerts by. TYPE: Union[PACategory, list[PACategory], None] DEFAULT: None
assignee	Assignee or list of uhashes to filter by. TYPE: Union[str, list, None] DEFAULT: None
created_from	Start of created date range (ISO or relative, e.g. -3d). TYPE: Optional[str] DEFAULT: None
created_until	End of created date range (ISO or relative). TYPE: Optional[str] DEFAULT: None
updated_from	Start of updated date range (ISO or relative). TYPE: Optional[str] DEFAULT: None
updated_until	End of updated date range (ISO or relative). TYPE: Optional[str] DEFAULT: None

Endpoints

• playbook-alert/search
• playbook-alert/{category}/{alert_id}

RAISES	DESCRIPTION
ValidationError	If any parameter is of incorrect type.
PlaybookAlertFetchError	If a connection or API error occurs.

RETURNS	DESCRIPTION
list[PLAYBOOK_ALERT_TYPE]	List of playbook alert ADTs matching the query or provided IDs.

Source code in psengine/playbook_alerts/playbook_alert_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=PlaybookAlertFetchError)
def fetch_bulk(
    self,
    alerts: Annotated[
        Optional[list[tuple[str, PACategory]]],
        Doc('List of (alert_id, category) tuples to fetch. Overrides search parameters.'),
    ] = None,
    panels: Annotated[
        Optional[list[str]],
        Doc('Panels to fetch for each alert. The `status` panel is always fetched.'),
    ] = None,
    fetch_images: Annotated[
        Optional[bool], Doc('Whether to fetch images for supported alert types.')
    ] = False,
    alerts_per_page: Annotated[
        Optional[int], Doc('Number of alerts per page (pagination).')
    ] = Field(ge=1, le=10000, default=ALERTS_PER_PAGE),
    max_results: Annotated[
        Optional[int], Doc('Maximum number of alerts to fetch.')
    ] = DEFAULT_LIMIT,
    order_by: Annotated[
        Optional[str], Doc('Field to order alerts by, e.g. `created` or `updated`.')
    ] = None,
    direction: Annotated[Optional[str], Doc('Sort direction: `asc` or `desc`.')] = None,
    entity: Annotated[
        Union[str, list, None], Doc('Entity or list of entities to filter alerts by.')
    ] = None,
    statuses: Annotated[
        Union[str, list, None],
        Doc("Status or list of statuses to filter alerts by, e.g. `['New', 'Closed']`."),
    ] = None,
    priority: Annotated[
        Union[str, list, None], Doc("Priority or list of priorities, e.g. `['High', 'Low']`.")
    ] = None,
    category: Annotated[
        Union[PACategory, list[PACategory], None],
        Doc('Category or list of categories to filter alerts by.'),
    ] = None,
    assignee: Annotated[
        Union[str, list, None], Doc('Assignee or list of uhashes to filter by.')
    ] = None,
    created_from: Annotated[
        Optional[str], Doc('Start of created date range (ISO or relative, e.g. `-3d`).')
    ] = None,
    created_until: Annotated[
        Optional[str], Doc('End of created date range (ISO or relative).')
    ] = None,
    updated_from: Annotated[
        Optional[str], Doc('Start of updated date range (ISO or relative).')
    ] = None,
    updated_until: Annotated[
        Optional[str], Doc('End of updated date range (ISO or relative).')
    ] = None,
) -> Annotated[
    list[PLAYBOOK_ALERT_TYPE],
    Doc('List of playbook alert ADTs matching the query or provided IDs.'),
]:
    """Fetch multiple playbook alerts in bulk, by query filters or specified alert tuples.

    Endpoints:
        - `playbook-alert/search`
        - `playbook-alert/{category}/{alert_id}`

    Raises:
        ValidationError: If any parameter is of incorrect type.
        PlaybookAlertFetchError: If a connection or API error occurs.
    """
    query_params = locals()
    for param in ['self', 'alerts', 'panels', 'fetch_images']:
        query_params.pop(param)
    if alerts is None:
        search_result = self.search(**query_params)
        alerts = [
            {'id': x.playbook_alert_id, 'category': x.category} for x in search_result.data
        ]
    else:
        alerts = [{'id': x[0], 'category': x[1]} for x in alerts]

    fetched_alerts = []
    errors = 0
    for cat in {x['category'] for x in alerts}:
        in_cat_alerts = filter(lambda x: x['category'] == cat, alerts)
        in_cat_ids = [x['id'] for x in in_cat_alerts]
        try:
            fetched_alerts.extend(self._do_bulk(in_cat_ids, cat, fetch_images, panels or []))
        except (PlaybookAlertBulkFetchError, PlaybookAlertRetrieveImageError) as err:  # noqa: PERF203
            errors += 1
            self.log.error(err)

    if errors:
        self.log.error(f'Failed to fetch alerts due to {errors} error(s). See errors above')
        raise PlaybookAlertFetchError('Failed to fetch alerts')

    return fetched_alerts

fetch_images

fetch_images(playbook_alert: PBA_WITH_IMAGES_TYPE) -> None

Retrieve images associated with a playbook alert, if available.

PARAMETER	DESCRIPTION
playbook_alert	A playbook alert ADT instance that supports image retrieval. TYPE: PBA_WITH_IMAGES_TYPE

Endpoints

• playbook-alert/domain_abuse/{alert_id}/image/{image_id}
• playbook-alert/geopolitics_facility/image/{image_id}

Example

Search and retrieve images for alerts:

from psengine.playbook_alerts import PlaybookAlertMgr, PBA_WITH_IMAGES_INST

mgr = PlaybookAlertMgr()
alerts = mgr.search()
alerts_to_fetch = [(a.playbook_alert_id, a.category) for a in alerts.data]

alerts_details = mgr.fetch_bulk(alerts_to_fetch)
retrieve_images_alerts = [
    a for a in alerts_details if isinstance(a, PBA_WITH_IMAGES_INST)
]

for alert in retrieve_images_alerts:
    mgr.fetch_images(alert)
    print(alert.images)

RAISES	DESCRIPTION
ValidationError	If the parameter is of incorrect type.
PlaybookAlertRetrieveImageError	If an API error occurs during image retrieval.

RETURNS	DESCRIPTION
None	This method modifies the input alert in-place by populating its images field.

Source code in psengine/playbook_alerts/playbook_alert_mgr.py

@debug_call
@validate_call
@connection_exceptions(
    ignore_status_code=[], exception_to_raise=PlaybookAlertRetrieveImageError
)
def fetch_images(
    self,
    playbook_alert: Annotated[
        PBA_WITH_IMAGES_TYPE,
        Doc('A playbook alert ADT instance that supports image retrieval.'),
    ],
) -> Annotated[
    None, Doc('This method modifies the input alert in-place by populating its `images` field.')
]:
    """Retrieve images associated with a playbook alert, if available.

    Endpoints:
        - `playbook-alert/domain_abuse/{alert_id}/image/{image_id}`
        - `playbook-alert/geopolitics_facility/image/{image_id}`

    Example:
        Search and retrieve images for alerts:

        ```python
        from psengine.playbook_alerts import PlaybookAlertMgr, PBA_WITH_IMAGES_INST

        mgr = PlaybookAlertMgr()
        alerts = mgr.search()
        alerts_to_fetch = [(a.playbook_alert_id, a.category) for a in alerts.data]

        alerts_details = mgr.fetch_bulk(alerts_to_fetch)
        retrieve_images_alerts = [
            a for a in alerts_details if isinstance(a, PBA_WITH_IMAGES_INST)
        ]

        for alert in retrieve_images_alerts:
            mgr.fetch_images(alert)
            print(alert.images)
        ```

    Raises:
        ValidationError: If the parameter is of incorrect type.
        PlaybookAlertRetrieveImageError: If an API error occurs during image retrieval.
    """
    for image_id in playbook_alert.image_ids:
        image_bytes = self.fetch_one_image(
            playbook_alert.playbook_alert_id, image_id, playbook_alert.category
        )
        playbook_alert.store_image(image_id, image_bytes)

fetch_one_image

fetch_one_image(
    alert_id: Optional[str] = None,
    image_id: Optional[str] = None,
    alert_category: PBA_WITH_IMAGES_VALIDATOR = 'domain_abuse',
) -> bytes

Retrieve an image from a playbook alert that includes visual content.

PARAMETER	DESCRIPTION
alert_id	Alert ID corresponding to the image ID. TYPE: Optional[str] DEFAULT: None
image_id	ID of the image to retrieve. TYPE: Optional[str] DEFAULT: None
alert_category	Category of the alert (e.g., 'domain_abuse', 'geopolitics_facility'). TYPE: PBA_WITH_IMAGES_VALIDATOR DEFAULT: 'domain_abuse'

Endpoints

• playbook-alert/domain_abuse/{alert_id}/image/{image_id}
• playbook-alert/geopolitics_facility/image/{image_id}

RAISES	DESCRIPTION
ValidationError	If any parameter is of incorrect type.
PlaybookAlertRetrieveImageError	If the image fetch request fails.

RETURNS	DESCRIPTION
bytes	Raw image content in bytes.

Source code in psengine/playbook_alerts/playbook_alert_mgr.py

@debug_call
@validate_call
@connection_exceptions(
    ignore_status_code=[], exception_to_raise=PlaybookAlertRetrieveImageError
)
def fetch_one_image(
    self,
    alert_id: Annotated[Optional[str], Doc('Alert ID corresponding to the image ID.')] = None,
    image_id: Annotated[Optional[str], Doc('ID of the image to retrieve.')] = None,
    alert_category: Annotated[
        PBA_WITH_IMAGES_VALIDATOR,
        Doc("Category of the alert (e.g., 'domain_abuse', 'geopolitics_facility')."),
    ] = 'domain_abuse',
) -> Annotated[bytes, Doc('Raw image content in bytes.')]:
    """Retrieve an image from a playbook alert that includes visual content.

    Endpoints:
        - `playbook-alert/domain_abuse/{alert_id}/image/{image_id}`
        - `playbook-alert/geopolitics_facility/image/{image_id}`

    Raises:
        ValidationError: If any parameter is of incorrect type.
        PlaybookAlertRetrieveImageError: If the image fetch request fails.
    """
    if alert_category == PACategory.DOMAIN_ABUSE.value:
        url = f'/{alert_category}/{alert_id}/image/{image_id}'
    else:
        url = f'/{alert_category}/image/{image_id}'

    self.log.info(f'Retrieving image: {image_id} for alert: {alert_id}')
    response = self.rf_client.request('get', EP_PLAYBOOK_ALERT + url)

    return response.content

search

search(
    alerts_per_page: Optional[int] = Field(
        ge=1, le=10000, default=ALERTS_PER_PAGE
    ),
    max_results: Optional[int] = DEFAULT_LIMIT,
    order_by: Optional[str] = None,
    direction: Optional[str] = None,
    entity: Union[str, list, None] = None,
    statuses: Union[str, list, None] = None,
    priority: Union[str, list, None] = None,
    category: Union[
        PACategory, list[PACategory], None
    ] = None,
    assignee: Union[str, list, None] = None,
    created_from: Optional[str] = None,
    created_until: Optional[str] = None,
    updated_from: Optional[str] = None,
    updated_until: Optional[str] = None,
) -> SearchResponse

Search for playbook alerts using filters.

PARAMETER	DESCRIPTION
alerts_per_page	Number of alerts per page. TYPE: Optional[int] DEFAULT: Field(ge=1, le=10000, default=ALERTS_PER_PAGE)
max_results	Maximum total number of alerts to fetch. TYPE: Optional[int] DEFAULT: DEFAULT_LIMIT
order_by	Field to order alerts by, e.g. created or updated. TYPE: Optional[str] DEFAULT: None
direction	Sort direction, either asc or desc. TYPE: Optional[str] DEFAULT: None
entity	Entity or list of entities to filter alerts by. TYPE: Union[str, list, None] DEFAULT: None
statuses	Status or list of statuses to filter alerts by. TYPE: Union[str, list, None] DEFAULT: None
priority	Priority or list of priorities to filter alerts by. TYPE: Union[str, list, None] DEFAULT: None
category	Category or list of categories to filter alerts by. TYPE: Union[PACategory, list[PACategory], None] DEFAULT: None
assignee	Assignee or list of assignees (uhashes) to filter alerts by. TYPE: Union[str, list, None] DEFAULT: None
created_from	Start of created date range (ISO or relative, e.g. -7d). TYPE: Optional[str] DEFAULT: None
created_until	End of created date range (ISO or relative). TYPE: Optional[str] DEFAULT: None
updated_from	Start of updated date range (ISO or relative). TYPE: Optional[str] DEFAULT: None
updated_until	End of updated date range (ISO or relative). TYPE: Optional[str] DEFAULT: None

Endpoint

playbook-alert/search

RAISES	DESCRIPTION
ValidationError	If any parameter is of incorrect type.
PlaybookAlertSearchError	If a connection or API error occurs.

RETURNS	DESCRIPTION
SearchResponse	Search results matching the alert query.

Source code in psengine/playbook_alerts/playbook_alert_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=PlaybookAlertSearchError)
def search(
    self,
    alerts_per_page: Annotated[Optional[int], Doc('Number of alerts per page.')] = Field(
        ge=1, le=10000, default=ALERTS_PER_PAGE
    ),
    max_results: Annotated[
        Optional[int], Doc('Maximum total number of alerts to fetch.')
    ] = DEFAULT_LIMIT,
    order_by: Annotated[
        Optional[str], Doc('Field to order alerts by, e.g. `created` or `updated`.')
    ] = None,
    direction: Annotated[Optional[str], Doc('Sort direction, either `asc` or `desc`.')] = None,
    entity: Annotated[
        Union[str, list, None], Doc('Entity or list of entities to filter alerts by.')
    ] = None,
    statuses: Annotated[
        Union[str, list, None], Doc('Status or list of statuses to filter alerts by.')
    ] = None,
    priority: Annotated[
        Union[str, list, None], Doc('Priority or list of priorities to filter alerts by.')
    ] = None,
    category: Annotated[
        Union[PACategory, list[PACategory], None],
        Doc('Category or list of categories to filter alerts by.'),
    ] = None,
    assignee: Annotated[
        Union[str, list, None],
        Doc('Assignee or list of assignees (uhashes) to filter alerts by.'),
    ] = None,
    created_from: Annotated[
        Optional[str], Doc('Start of created date range (ISO or relative, e.g. `-7d`).')
    ] = None,
    created_until: Annotated[
        Optional[str], Doc('End of created date range (ISO or relative).')
    ] = None,
    updated_from: Annotated[
        Optional[str], Doc('Start of updated date range (ISO or relative).')
    ] = None,
    updated_until: Annotated[
        Optional[str], Doc('End of updated date range (ISO or relative).')
    ] = None,
) -> Annotated[SearchResponse, Doc('Search results matching the alert query.')]:
    """Search for playbook alerts using filters.

    Endpoint:
        `playbook-alert/search`

    Raises:
        ValidationError: If any parameter is of incorrect type.
        PlaybookAlertSearchError: If a connection or API error occurs.
    """
    query_params = locals()
    query_params.pop('self')
    request_body = self._prepare_query(**query_params).json()
    self.log.info(
        f'Searching for playbook alert query: {request_body}, max_results: {max_results}'
    )

    search_results = self.rf_client.request_paged(
        method='post',
        url=EP_PLAYBOOK_ALERT_SEARCH,
        data=request_body,
        max_results=max_results,
        results_path='data',
        offset_key='from',
    )

    # To avoid a breaking change have to reconstruct the SearchResponse model manually
    # We did lost the total count the API "could" return
    result = {
        'status': {'status_code': 'Ok', 'status_message': 'Playbook alert search successful'},
        'data': search_results,
        'counts': {'returned': len(search_results), 'total': len(search_results)},
    }

    self.log.info(f'Search returned {len(search_results)} playbook alerts')

    return SearchResponse.model_validate(result)

update

update(
    alert: Union[PLAYBOOK_ALERT_TYPE, str],
    priority: Optional[str] = None,
    status: Optional[str] = None,
    assignee: Optional[str] = None,
    log_entry: Optional[str] = None,
    reopen_strategy: Optional[str] = None,
) -> Response

Update a playbook alert.

PARAMETER	DESCRIPTION
alert	Playbook alert ADT or alert ID to update. TYPE: Union[PLAYBOOK_ALERT_TYPE, str]
priority	Updated alert priority (e.g. 'High', 'Low'). TYPE: Optional[str] DEFAULT: None
status	Updated alert status (e.g. 'New', 'InProgress'). TYPE: Optional[str] DEFAULT: None
assignee	Assignee uhash for the alert. TYPE: Optional[str] DEFAULT: None
log_entry	Text for the alert log entry. TYPE: Optional[str] DEFAULT: None
reopen_strategy	Strategy for reopening closed alerts. TYPE: Optional[str] DEFAULT: None

Endpoint

playbook-alert/common/{playbook_alert_id}

RAISES	DESCRIPTION
ValidationError	If any parameter is of incorrect type.
ValueError	If no update parameters are provided.
PlaybookAlertUpdateError	If the update request fails.

RETURNS	DESCRIPTION
Response	API response object for the update operation.

Source code in psengine/playbook_alerts/playbook_alert_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=PlaybookAlertUpdateError)
def update(
    self,
    alert: Annotated[
        Union[PLAYBOOK_ALERT_TYPE, str], Doc('Playbook alert ADT or alert ID to update.')
    ],
    priority: Annotated[
        Optional[str], Doc("Updated alert priority (e.g. 'High', 'Low').")
    ] = None,
    status: Annotated[
        Optional[str], Doc("Updated alert status (e.g. 'New', 'InProgress').")
    ] = None,
    assignee: Annotated[Optional[str], Doc('Assignee uhash for the alert.')] = None,
    log_entry: Annotated[Optional[str], Doc('Text for the alert log entry.')] = None,
    reopen_strategy: Annotated[
        Optional[str], Doc('Strategy for reopening closed alerts.')
    ] = None,
) -> Annotated[requests.Response, Doc('API response object for the update operation.')]:
    """Update a playbook alert.

    Endpoint:
        `playbook-alert/common/{playbook_alert_id}`

    Raises:
        ValidationError: If any parameter is of incorrect type.
        ValueError: If no update parameters are provided.
        PlaybookAlertUpdateError: If the update request fails.
    """
    body = {
        'priority': priority,
        'status': status,
        'assignee': assignee,
        'log_entry': log_entry,
        'reopen': reopen_strategy,
    }

    body = {k: v for k, v in body.items() if v is not None}
    if not body:
        raise ValueError('No update parameters were supplied')

    alert_id = alert.playbook_alert_id if isinstance(alert, PLAYBOOK_ALERT_INST) else alert
    validated_payload = UpdateAlertIn.model_validate(body)

    url = f'{EP_PLAYBOOK_ALERT_COMMON}/{alert_id}'
    self.log.info(f'Updating playbook alert: {alert_id}')

    return self.rf_client.request('put', url=url, data=validated_payload.json())