forked from keephq/keep
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Used pagertree REST API to add agertree provider. Signed-off-by: Jay Kumar <[email protected]>
- Loading branch information
Showing
6 changed files
with
282 additions
and
13 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| --- | ||
| title: "Pagertree Provider" | ||
| description: "The Pagertree Provider facilitates interactions with the Pagertree API, allowing the retrieval and management of alerts." | ||
| --- | ||
|
|
||
| ## Inputs | ||
|
|
||
| The `notify` function in the `PagertreeProvider` class takes the following parameters: | ||
|
|
||
| ```python | ||
| kwargs(dict): | ||
| title (str): Title of the alert or incident. *Required* | ||
| urgency (Literal["low", "medium", "high", "critical"]): Defines the urgency of the alert. *Required* | ||
| incident (bool, default=False): If True, sends data as an incident. *Optional* | ||
| severities (Literal["SEV-1", "SEV-2", "SEV-3", "SEV-4", "SEV-5", "SEV_UNKNOWN"], default="SEV-5"): Specifies the severity level of the incident. *Optional* | ||
| incident_message (str, default=""): Message describing the incident. *Optional* | ||
| description (str, default=""): Detailed description of the alert or incident. *Optional* | ||
| status (Literal["queued", "open", "acknowledged", "resolved", "dropped"], default="queued"): Status of the alert or incident. *Optional* | ||
| destination_team_ids (list[str], default=[]): List of team IDs that the alert or incident will be sent to. *Optional* | ||
| destination_router_ids (list[str], default=[]): List of router IDs that the alert or incident will be sent to. *Optional* | ||
| destination_account_user_ids (list[str], default=[]): List of account user IDs that the alert or incident will be sent to. *Optional* | ||
| **kwargs (dict): Additional keyword arguments that might be needed for future use. *Optional* | ||
| ``` | ||
|
|
||
|
|
||
| ### Authentication Parameters | ||
|
|
||
| The `PagertreeProviderAuthConfig` class takes the following parameters: | ||
| - api_token (str): Your Pagertree API Token. *Required* | ||
|
|
||
|
|
||
| ## Connecting with the Provider | ||
|
|
||
| - To interact with the Pagertree API, you need to provide an api_token. | ||
| - You can view and manage your API keys on your [User Settings](https://app.pagertree.com/user/settings) page. | ||
|
|
||
|
|
||
| ## Notes | ||
|
|
||
| _This provider uses the Pagertree API to send alerts or mark them as incidents based on the parameters provided. Depending on whether an incident is flagged as true, it either calls `__send_alert` or `__send_incident` method._ | ||
|
|
||
|
|
||
| ## Useful Links | ||
|
|
||
| - Pagertree API documentation: [Pagertree API](https://pagertree.com/docs) | ||
| - Pagertree Authentication: [Authentication](https://pagertree.com/docs/api/authentication) | ||
| - Pagertree Alerts: [Alerts & Incident](https://pagertree.com/docs/api/alerts) |
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Empty file.
202 changes: 202 additions & 0 deletions
202
keep/providers/pagertree_provider/pagertree_provider.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,202 @@ | ||
| """ | ||
| PagetreeProvider is a class that provides a way to read get alerts from Pagetree. | ||
| """ | ||
|
|
||
| import dataclasses | ||
| from typing import Literal | ||
|
|
||
| import pydantic | ||
| import requests | ||
|
|
||
| from keep.api.models.alert import AlertDto | ||
| from keep.contextmanager.contextmanager import ContextManager | ||
| from keep.providers.base.base_provider import BaseProvider | ||
| from keep.providers.models.provider_config import ProviderConfig, ProviderScope | ||
|
|
||
|
|
||
| @pydantic.dataclasses.dataclass | ||
| class PagertreeProviderAuthConfig: | ||
| api_token: str = dataclasses.field( | ||
| metadata={ | ||
| "required": True, | ||
| "description": "Your pagertree APIToken", | ||
| "sensitive": True, | ||
| }, | ||
| default=None, | ||
| ) | ||
|
|
||
|
|
||
| class PagertreeProvider(BaseProvider): | ||
| """Get all alerts from pagertree""" | ||
|
|
||
| PROVIDER_DISPLAY_NAME = "pagertree" | ||
|
|
||
| PROVIDER_SCOPES = [ | ||
| ProviderScope( | ||
| name="authenticated", | ||
| description="The user can connect to the server and is authenticated using their API_Key", | ||
| mandatory=True, | ||
| alias="Authenticated with pagertree", | ||
| ) | ||
| ] | ||
|
|
||
| def __init__( | ||
| self, context_manager: ContextManager, provider_id: str, config: ProviderConfig | ||
| ): | ||
| super().__init__(context_manager, provider_id, config) | ||
|
|
||
| def __get_headers(self): | ||
| return { | ||
| 'Accept': 'application/json', | ||
| 'Authorization': f'Bearer {self.authentication_config.api_token}', | ||
| } | ||
|
|
||
| def validate_scopes(self): | ||
| """ | ||
| Validates that the user has the required scopes to use the provider. | ||
| """ | ||
| try: | ||
| response = requests.get('https://api.pagertree.com/api/v4/alerts', headers=self.__get_headers()) | ||
|
|
||
| if response.status_code == 200: | ||
| scopes = { | ||
| "authenticated": True, | ||
| } | ||
| else: | ||
| self.logger.error("Unable to authenticate user") | ||
| scopes = { | ||
| "authenticated": f"User not authorized, StatusCode: {response.status_code}", | ||
| } | ||
| except Exception as e: | ||
| self.logger.error("Error validating scopes", extra={"error": str(e)}) | ||
| scopes = { | ||
| "authenticated": str(e), | ||
| } | ||
| return scopes | ||
|
|
||
| def dispose(self): | ||
| pass | ||
|
|
||
| def validate_config(self): | ||
| """ | ||
| Validates required configuration for pgartree's provider. | ||
| """ | ||
| self.authentication_config = PagertreeProviderAuthConfig( | ||
| **self.config.authentication | ||
| ) | ||
|
|
||
| def _get_alerts(self) -> list[AlertDto]: | ||
| try: | ||
| response = requests.get('https://api.pagertree.com/api/v4/alerts', headers=self.__get_headers()) | ||
| if not response.ok: | ||
| self.logger.error("Failed to get alerts", extra=response.json()) | ||
| raise Exception("Could not get alerts") | ||
| return [AlertDto( | ||
| id=alert["id"], | ||
| status=alert["status"], | ||
| severity=alert["urgency"], | ||
| source=alert["source"], | ||
| message=alert["title"], | ||
| startedAt=alert["created_at"], | ||
| description=alert["description"] | ||
| ) for alert in response.json()['alerts']] | ||
|
|
||
| except Exception as e: | ||
| self.logger.error("Error while getting PagerTree alerts", extra={"error": str(e)}) | ||
| raise e | ||
|
|
||
| def __send_alert(self, | ||
| title: str, | ||
| description: str, | ||
| urgency: Literal["low", "medium", "high", "critical"], | ||
| destination_team_ids: list[str], | ||
| destination_router_ids: list[str], | ||
| destination_account_user_ids: list[str], | ||
| status: Literal["queued", "open", "acknowledged", "resolved", "dropped"], | ||
| **kwargs: dict, ): | ||
| """ | ||
| Sends PagerDuty Alert | ||
| Args: | ||
| title: Title of the alert. | ||
| description: UTF-8 string of custom message for alert. Shown in incident description | ||
| urgency: low|medium|high|critical | ||
| destination_team_ids: destination team_ids to send alert to | ||
| destination_router_ids: destination router_ids to send alert to | ||
| destination_account_user_ids: destination account_users_ids to send alert to | ||
| status: alert status to send | ||
| """ | ||
| response = requests.post('https://api.pagertree.com/api/v4/alerts', headers=self.__get_headers(), data={ | ||
| "title": title, | ||
| "description": description, | ||
| "urgency": urgency, | ||
| "destination_team_ids": destination_team_ids, | ||
| "destination_router_ids": destination_router_ids, | ||
| "destination_account_user_ids": destination_account_user_ids, | ||
| "status": status, | ||
| **kwargs | ||
| }) | ||
| if not response.ok: | ||
| self.logger.error("Failed to send alert", extra={"error": response.json()}) | ||
| self.logger.info("Alert status: %s", response.status_code) | ||
| self.logger.info("Alert created successfully", response.json()) | ||
|
|
||
| def __send_incident(self, title: str, | ||
| incident_severity: str, | ||
| incident_message: str, | ||
| urgency: Literal["low", "medium", "high", "critical"], | ||
| destination_team_ids: list[str], | ||
| destination_router_ids: list[str], | ||
| destination_account_user_ids: list[str], | ||
| **kwargs: dict, ): | ||
| """ | ||
| Marking an alert as an incident communicates to your team members this alert is a greater degree of severity than a normal alert. | ||
| Args: | ||
| title: Title of the alert. | ||
| description: UTF-8 string of custom message for alert. Shown in incident description | ||
| urgency: low|medium|high|critical | ||
| destination_team_ids: destination team_ids to send alert to | ||
| destination_router_ids: destination router_ids to send alert to | ||
| destination_account_user_ids: destination account_users_ids to send alert to | ||
| """ | ||
| response = requests.post('https://api.pagertree.com/api/v4/alerts', headers=self.__get_headers(), data={ | ||
| "title": title, | ||
| "meta": { | ||
| "incident": True, | ||
| "incident_severity": incident_severity, | ||
| "incident_message": incident_message | ||
| }, | ||
| "urgency": urgency, | ||
| "destination_team_ids": destination_team_ids, | ||
| "destination_router_ids": destination_router_ids, | ||
| "destination_account_user_ids": destination_account_user_ids, | ||
| **kwargs | ||
| }) | ||
| if not response.ok: | ||
| self.logger.error("Failed to send incident", extra={"error": response.json()}) | ||
| self.logger.info("Incident status: %s", response.status_code) | ||
| self.logger.info("Incident created successfully", response.json()) | ||
|
|
||
| def notify(self, | ||
| title: str, | ||
| urgency: Literal["low", "medium", "high", "critical"], | ||
| incident: bool = False, | ||
| severities: Literal["SEV-1", "SEV-2", "SEV-3", "SEV-4", "SEV-5", "SEV_UNKNOWN"] = "SEV-5", | ||
| incident_message: str = "", | ||
| description: str = "", | ||
| status: Literal["queued", "open", "acknowledged", "resolved", "dropped"] = "queued", | ||
| destination_team_ids: list[str] = [], | ||
| destination_router_ids: list[str] = [], | ||
| destination_account_user_ids: list[str] = [], | ||
| **kwargs: dict, ): | ||
| if len(destination_team_ids) + len(destination_router_ids) + len(destination_account_user_ids) == 0: | ||
| raise Exception("at least 1 destination (Team, Router, or Account User) is required") | ||
| if not incident: | ||
| self.__send_alert(title, description, urgency, destination_team_ids, destination_router_ids, | ||
| destination_account_user_ids, status, **kwargs) | ||
| else: | ||
| self.__send_incident(incident_message, severities, title, urgency, destination_team_ids, | ||
| destination_router_ids, | ||
| destination_account_user_ids, **kwargs) |