Skip to content

Commit

Permalink
feat: Adjust the alerting model
Browse files Browse the repository at this point in the history
* feat: Add redirect support
* fix: Adjust the alerting functionality
  • Loading branch information
ZPascal committed Dec 22, 2024
1 parent 161dfd5 commit 94f9758
Show file tree
Hide file tree
Showing 6 changed files with 104 additions and 111 deletions.
14 changes: 7 additions & 7 deletions docs/content/grafana_api/alerting.md
Original file line number Diff line number Diff line change
Expand Up @@ -595,14 +595,14 @@ The method includes a functionality to get a ruler group specified by the namesp
#### test\_rule

```python
def test_rule(data_query: list) -> dict
def test_rule(data_queries: list) -> dict
```

The method includes a functionality to test a rule specified by a list of datasource rule query objects

**Arguments**:

- `data_query` _list_ - Specify a list of datasource rule query objects
- `data_queries` _list_ - Specify a list of datasource rule query objects


**Raises**:
Expand All @@ -622,7 +622,7 @@ The method includes a functionality to test a rule specified by a list of dataso
```python
def test_datasource_uid_rule(expr: str,
condition: str,
data_query: list,
data_queries: list,
datasource_uid: str = "grafana") -> dict
```

Expand All @@ -632,7 +632,7 @@ The method includes a functionality to test a datasource uid rule specified by t

- `expr` _str_ - Specify a list of datasource rule query objects
- `condition` _str_ - Specify the condition
- `data_query` _list_ - Specify a list of datasource rule query objects
- `data_queries` _list_ - Specify a list of datasource rule query objects
- `datasource_uid` _str_ - Specify the datasource uid or recipient of the alerts (default grafana)


Expand All @@ -651,15 +651,15 @@ The method includes a functionality to test a datasource uid rule specified by t
#### test\_backtest\_rule

```python
def test_backtest_rule(condition: str, data_query: list) -> dict
def test_backtest_rule(condition: str, data_queries: list) -> dict
```

The method includes a functionality to test a rule specified by the condition, a list of data queries
The method includes a functionality to test a rule specified by the condition and a list of data queries

**Arguments**:

- `condition` _str_ - Specify the condition
- `data_query` _list_ - Specify a list of datasource rule query objects
- `data_queries` _list_ - Specify a list of datasource rule query objects


**Raises**:
Expand Down
26 changes: 14 additions & 12 deletions docs/content/grafana_api/model.md
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,7 @@ The class includes all necessary variables to establish a connection to the Graf
- `ssl_context` _ssl.SSLContext_ - Specify the custom ssl context of the Grafana system
- `num_pools` _int_ - Specify the number of the connection pool
- `retries` _any_ - Specify the number of the retries. Please use False as parameter to disable the retries
- `follow_redirects` _bool_ - Specify if redirections should be followed (default True)

<a id="model.DatasourceQuery"></a>

Expand Down Expand Up @@ -421,12 +422,12 @@ The class includes all necessary variables to generate an Alertmanager config ob

**Arguments**:

- `global_config` _dict_ - Specify the global config of the Alertmanager
- `inhibit_rules` _list_ - Specify the inhibit rules of the Alertmanager
- `mute_time_intervals` _list_ - Specify the mute time intervals of the Alertmanager
- `global_config` _Union[dict, None]_ - Specify the global config of the Alertmanager
- `inhibit_rules` _Union[list, None]_ - Specify the inhibit rules of the Alertmanager
- `mute_time_intervals` _Union[list, None]_ - Specify the mute time intervals of the Alertmanager
- `receivers` _list_ - Specify the receiver's of the Alertmanager
- `route` _dict_ - Specify the route of the Alertmanager
- `templates` _list_ - Specify an Alertmanager template
- `templates` _Union[list, None]_ - Specify an Alertmanager template

<a id="model.AlertmanagerReceivers"></a>

Expand All @@ -442,15 +443,16 @@ The class includes all necessary variables to generate an Alertmanager receiver'
**Arguments**:

- `name` _str_ - Specify the name of the receiver
- `email_configs` _list_ - Specify the email configuration of the receiver's
- `email_configs` _Union[list, None]_ - Specify the email configuration of the receiver's
- `grafana_managed_receiver_configs` _list_ - Specify the Grafana managed receiver configuration of the receiver's
- `opsgenie_configs` _list_ - Specify the ops genie configuration of the receiver's
- `pagerduty_configs` _dict_ - Specify the pager duty configuration of the receiver's
- `pushover_configs` _list_ - Specify the push over configuration of the receiver's
- `slack_configs` _list_ - Specify the Slack configuration of the receiver's
- `victorops_configs` _list_ - Specify the victor ops configuration of the receiver's
- `webhook_configs` _list_ - Specify the webhook configuration of the receiver's
- `wechat_configs` _list_ - Specify the wechaty configuration of the receiver's
- `opsgenie_configs` _Union[list, None]_ - Specify the ops genie configuration of the receiver's
- `pagerduty_configs` _Union[list, None]_ - Specify the pager duty configuration of the receiver's
- `pushover_configs` _Union[list, None]_ - Specify the push over configuration of the receiver's
- `slack_configs` _Union[list, None]_ - Specify the Slack configuration of the receiver's
- `sns_configs` _Union[list, None]_ - Specify the SNS configuration of the receiver's
- `victorops_configs` _Union[list, None]_ - Specify the victor ops configuration of the receiver's
- `webhook_configs` _Union[list, None]_ - Specify the webhook configuration of the receiver's
- `wechat_configs` _Union[list, None]_ - Specify the wechaty configuration of the receiver's

<a id="model.RulerRule"></a>

Expand Down
34 changes: 17 additions & 17 deletions grafana_api/alerting.py
Original file line number Diff line number Diff line change
Expand Up @@ -773,11 +773,11 @@ def get_ruler_group(
)
raise ValueError

def test_rule(self, data_query: list) -> dict:
def test_rule(self, data_queries: list) -> dict:
"""The method includes a functionality to test a rule specified by a list of datasource rule query objects
Args:
data_query (list): Specify a list of datasource rule query objects
data_queries (list): Specify a list of datasource rule query objects
Raises:
ValueError: Missed specifying a necessary value
Expand All @@ -787,11 +787,11 @@ def test_rule(self, data_query: list) -> dict:
api_call (str): Returns the result of the specified query
"""

if data_query != list():
if data_queries != list():
datasource_rule_query_objects_json: list = list()
datasource_rule_query_object_json: dict = dict()

for datasource_rule_query_object in data_query:
for datasource_rule_query_object in data_queries:
datasource_rule_query_object_json["datasourceUid"] = (
datasource_rule_query_object.datasource_uid
)
Expand Down Expand Up @@ -828,22 +828,22 @@ def test_rule(self, data_query: list) -> dict:
else:
return api_call
else:
logging.error("There is no data_query defined.")
logging.error("There is no data_queries defined.")
raise ValueError

def test_datasource_uid_rule(
self,
expr: str,
condition: str,
data_query: list,
data_queries: list,
datasource_uid: str = "grafana",
) -> dict:
"""The method includes a functionality to test a datasource uid rule specified by the expr, the condition, a list of data queries and the datasource_uid
Args:
expr (str): Specify a list of datasource rule query objects
condition (str): Specify the condition
data_query (list): Specify a list of datasource rule query objects
data_queries (list): Specify a list of datasource rule query objects
datasource_uid (str): Specify the datasource uid or recipient of the alerts (default grafana)
Raises:
Expand All @@ -858,12 +858,12 @@ def test_datasource_uid_rule(
len(datasource_uid) != 0
and len(expr) != 0
and len(condition) != 0
and data_query != list()
and data_queries != list()
):
datasource_rule_query_objects_json: list = list()
datasource_rule_query_object_json: dict = dict()

for datasource_rule_query_object in data_query:
for datasource_rule_query_object in data_queries:
datasource_rule_query_object_json["datasourceUid"] = (
datasource_rule_query_object.datasource_uid
)
Expand All @@ -889,7 +889,7 @@ def test_datasource_uid_rule(
json.dumps(
{
"expr": expr,
"grafana_alert": {
"grafana_condition": {
"condition": condition,
"data": datasource_rule_query_objects_json,
"now": str(datetime.datetime.now()),
Expand All @@ -905,16 +905,16 @@ def test_datasource_uid_rule(
return api_call
else:
logging.error(
"There is no datasource_uid, expr, condition or data_query defined."
"There is no datasource_uid, expr, condition or data_queries defined."
)
raise ValueError

def test_backtest_rule(self, condition: str, data_query: list) -> dict:
"""The method includes a functionality to test a rule specified by the condition, a list of data queries
def test_backtest_rule(self, condition: str, data_queries: list) -> dict:
"""The method includes a functionality to test a rule specified by the condition and a list of data queries
Args:
condition (str): Specify the condition
data_query (list): Specify a list of datasource rule query objects
data_queries (list): Specify a list of datasource rule query objects
Raises:
ValueError: Missed specifying a necessary value
Expand All @@ -924,11 +924,11 @@ def test_backtest_rule(self, condition: str, data_query: list) -> dict:
api_call (dict): Returns the result of the specified rule
"""

if len(condition) != 0 and data_query != list():
if len(condition) != 0 and data_queries != list():
datasource_rule_query_objects_json: list = list()
datasource_rule_query_object_json: dict = dict()

for datasource_rule_query_object in data_query:
for datasource_rule_query_object in data_queries:
datasource_rule_query_object_json["datasourceUid"] = (
datasource_rule_query_object.datasource_uid
)
Expand Down Expand Up @@ -966,7 +966,7 @@ def test_backtest_rule(self, condition: str, data_query: list) -> dict:
return api_call
else:
logging.error(
"There is no datasource_uid, condition or data_query defined."
"There is no condition or data_queries defined."
)
raise ValueError

Expand Down
2 changes: 2 additions & 0 deletions grafana_api/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -343,6 +343,7 @@ def create_the_http_api_client(
headers=headers,
transport=async_transport,
verify=self.grafana_api_model.ssl_context,
follow_redirects=self.grafana_api_model.follow_redirects,
)
else:
return httpx.Client(
Expand All @@ -351,4 +352,5 @@ def create_the_http_api_client(
headers=headers,
transport=transport,
verify=self.grafana_api_model.ssl_context,
follow_redirects=self.grafana_api_model.follow_redirects,
)
53 changes: 28 additions & 25 deletions grafana_api/model.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@ class APIModel:
ssl_context (ssl.SSLContext): Specify the custom ssl context of the Grafana system
num_pools (int): Specify the number of the connection pool
retries (any): Specify the number of the retries. Please use False as parameter to disable the retries
follow_redirects (bool): Specify if redirections should be followed (default True)
"""

host: str
Expand All @@ -98,6 +99,7 @@ class APIModel:
ssl_context: ssl.SSLContext = httpx.create_ssl_context()
num_pools: int = 10
retries: any = 10
follow_redirects: bool = True


@dataclass
Expand Down Expand Up @@ -438,20 +440,20 @@ class AlertmanagerConfig:
"""The class includes all necessary variables to generate an Alertmanager config object that is necessary to communicate and set up the Grafana Alertmanager endpoint
Args:
global_config (dict): Specify the global config of the Alertmanager
inhibit_rules (list): Specify the inhibit rules of the Alertmanager
mute_time_intervals (list): Specify the mute time intervals of the Alertmanager
global_config (Union[dict, None]): Specify the global config of the Alertmanager
inhibit_rules (Union[list, None]): Specify the inhibit rules of the Alertmanager
mute_time_intervals (Union[list, None]): Specify the mute time intervals of the Alertmanager
receivers (list): Specify the receiver's of the Alertmanager
route (dict): Specify the route of the Alertmanager
templates (list): Specify an Alertmanager template
templates (Union[list, None]): Specify an Alertmanager template
"""

global_config: dict
inhibit_rules: list
mute_time_intervals: list
global_config: Union[dict, None]
inhibit_rules: Union[list, None]
mute_time_intervals: Union[list, None]
receivers: list
route: dict
templates: list
templates: Union[list, None]


@dataclass
Expand All @@ -460,28 +462,29 @@ class AlertmanagerReceivers:
Args:
name (str): Specify the name of the receiver
email_configs (list): Specify the email configuration of the receiver's
email_configs (Union[list, None]): Specify the email configuration of the receiver's
grafana_managed_receiver_configs (list): Specify the Grafana managed receiver configuration of the receiver's
opsgenie_configs (list): Specify the ops genie configuration of the receiver's
pagerduty_configs (dict): Specify the pager duty configuration of the receiver's
pushover_configs (list): Specify the push over configuration of the receiver's
slack_configs (list): Specify the Slack configuration of the receiver's
victorops_configs (list): Specify the victor ops configuration of the receiver's
webhook_configs (list): Specify the webhook configuration of the receiver's
wechat_configs (list): Specify the wechaty configuration of the receiver's
opsgenie_configs (Union[list, None]): Specify the ops genie configuration of the receiver's
pagerduty_configs (Union[list, None]): Specify the pager duty configuration of the receiver's
pushover_configs (Union[list, None]): Specify the push over configuration of the receiver's
slack_configs (Union[list, None]): Specify the Slack configuration of the receiver's
sns_configs (Union[list, None]): Specify the SNS configuration of the receiver's
victorops_configs (Union[list, None]): Specify the victor ops configuration of the receiver's
webhook_configs (Union[list, None]): Specify the webhook configuration of the receiver's
wechat_configs (Union[list, None]): Specify the wechaty configuration of the receiver's
"""

name: str
email_configs: list
email_configs: Union[list, None]
grafana_managed_receiver_configs: list
opsgenie_configs: list
pagerduty_configs: list
pushover_configs: list
slack_configs: list
sns_configs: list
victorops_configs: list
webhook_configs: list
wechat_configs: list
opsgenie_configs: Union[list, None]
pagerduty_configs: Union[list, None]
pushover_configs: Union[list, None]
slack_configs: Union[list, None]
sns_configs: Union[list, None]
victorops_configs: Union[list, None]
webhook_configs: Union[list, None]
wechat_configs: Union[list, None]


@dataclass
Expand Down
Loading

0 comments on commit 94f9758

Please sign in to comment.