"""Async Kibana Alerting API client.
Covers the Kibana 9.4.3 Alerting HTTP APIs: rule CRUD and lifecycle
operations (``client.alerting.rule``), backfill scheduling
(``client.alerting.backfill``), plus framework-level helpers
(``client.alerting.health()`` and ``client.alerting.rule_types()``).
"""
from __future__ import annotations
import json
from typing import TYPE_CHECKING, Any
from elastic_transport import ObjectApiResponse
from kibana._async.client.utils import AsyncNamespaceClient, _quote
if TYPE_CHECKING:
from kibana._async.client import AsyncKibana
[docs]
class AsyncRulesClient(AsyncNamespaceClient):
"""Async client for Kibana Alerting rule operations.
Provides CRUD plus lifecycle operations (enable/disable, mute/unmute,
snooze/unsnooze, API-key rotation) for alerting rules.
Example:
>>> from kibana import AsyncKibana
>>> client = AsyncKibana("http://localhost:5601", api_key="...")
>>> rule = await client.alerting.rule.create(
... name="CPU Alert",
... consumer="alerts",
... rule_type_id=".index-threshold",
... schedule={"interval": "1m"},
... params={"index": ["logs-*"], "timeField": "@timestamp"},
... )
"""
[docs]
async def create(
self,
*,
name: str,
consumer: str,
rule_type_id: str,
schedule: dict[str, Any],
params: dict[str, Any],
id: str | None = None,
actions: list[dict[str, Any]] | None = None,
tags: list[str] | None = None,
notify_when: str | None = None,
enabled: bool = True,
throttle: str | None = None,
alert_delay: dict[str, Any] | None = None,
flapping: dict[str, Any] | None = None,
artifacts: dict[str, Any] | None = None,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Create a new rule.
Calls ``POST /api/alerting/rule`` (or ``POST /api/alerting/rule/{id}``
when a caller-chosen ``id`` is given).
Args:
name: Human-readable name for the rule.
consumer: The application or feature that owns the rule
(e.g. ``alerts``, ``siem``, ``stackAlerts``).
rule_type_id: The rule type ID (e.g. ``.es-query``).
schedule: Check interval, e.g. ``{"interval": "1m"}``.
params: Rule-type specific parameters.
id: Optional caller-chosen rule ID. If omitted, an ID is
randomly generated by Kibana.
actions: List of actions to run when the rule triggers.
tags: List of tags to organize the rule.
notify_when: When to notify (e.g. ``onActionGroupChange``,
``onActiveAlert``, ``onThrottleInterval``).
enabled: Whether the rule is enabled on creation.
throttle: Deprecated in favor of the action ``frequency`` object.
How long to wait between actions (e.g. ``1m``).
alert_delay: Number of consecutive runs that must meet the rule
conditions before an alert occurs, e.g. ``{"active": 3}``.
flapping: Flapping detection settings, e.g.
``{"look_back_window": 10, "status_change_threshold": 3}``.
artifacts: Rule artifacts (e.g. dashboards and investigation
guide) attached to the rule.
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
The created rule.
Raises:
ValueError: If a required parameter is missing.
BadRequestError: If the rule configuration is invalid.
Example:
>>> await client.alerting.rule.create(
... name="ES query rule",
... consumer="alerts",
... rule_type_id=".es-query",
... schedule={"interval": "1m"},
... params={"searchType": "esQuery", "size": 1},
... id="my-rule-id",
... )
"""
if not name:
raise ValueError("Parameter 'name' is required")
if not consumer:
raise ValueError("Parameter 'consumer' is required")
if not rule_type_id:
raise ValueError("Parameter 'rule_type_id' is required")
if schedule is None:
raise ValueError("Parameter 'schedule' is required")
if params is None:
raise ValueError("Parameter 'params' is required")
body: dict[str, Any] = {
"name": name,
"consumer": consumer,
"rule_type_id": rule_type_id,
"schedule": schedule,
"params": params,
"enabled": enabled,
}
if actions is not None:
body["actions"] = actions
if tags is not None:
body["tags"] = tags
if notify_when is not None:
body["notify_when"] = notify_when
if throttle is not None:
body["throttle"] = throttle
if alert_delay is not None:
body["alert_delay"] = alert_delay
if flapping is not None:
body["flapping"] = flapping
if artifacts is not None:
body["artifacts"] = artifacts
base_path = "/api/alerting/rule"
if id is not None:
base_path = f"{base_path}/{_quote(id)}"
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path(base_path, space_id)
return await self.perform_request("POST", path, body=body)
[docs]
async def get(
self,
*,
id: str,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Retrieve a rule by ID.
Calls ``GET /api/alerting/rule/{id}``.
Args:
id: Rule ID.
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
The rule object.
Raises:
ValueError: If ``id`` is missing.
NotFoundError: If the rule does not exist.
Example:
>>> rule = await client.alerting.rule.get(id="my-rule-id")
>>> rule["name"]
"""
if not id:
raise ValueError("Parameter 'id' is required")
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path(f"/api/alerting/rule/{_quote(id)}", space_id)
return await self.perform_request("GET", path)
[docs]
async def update(
self,
*,
id: str,
name: str,
schedule: dict[str, Any],
params: dict[str, Any] | None = None,
actions: list[dict[str, Any]] | None = None,
tags: list[str] | None = None,
notify_when: str | None = None,
throttle: str | None = None,
alert_delay: dict[str, Any] | None = None,
flapping: dict[str, Any] | None = None,
artifacts: dict[str, Any] | None = None,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Update an existing rule.
Calls ``PUT /api/alerting/rule/{id}``. Note: ``rule_type_id``,
``consumer`` and ``enabled`` are immutable after creation; use the
enable/disable API to change the enabled state.
Args:
id: Rule ID to update.
name: Human-readable name for the rule.
schedule: Check interval, e.g. ``{"interval": "1m"}``.
params: Rule-type specific parameters.
actions: List of actions to run when the rule triggers.
tags: List of tags to organize the rule.
notify_when: When to notify.
throttle: Deprecated in favor of the action ``frequency`` object.
How long to wait between actions.
alert_delay: Number of consecutive runs that must meet the rule
conditions before an alert occurs, e.g. ``{"active": 3}``.
flapping: Flapping detection settings.
artifacts: Rule artifacts attached to the rule.
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
The updated rule.
Raises:
ValueError: If ``id`` is missing.
NotFoundError: If the rule does not exist.
Example:
>>> await client.alerting.rule.update(
... id="my-rule-id",
... name="Renamed rule",
... schedule={"interval": "5m"},
... params={"searchType": "esQuery", "size": 1},
... )
"""
if not id:
raise ValueError("Parameter 'id' is required")
body: dict[str, Any] = {
"name": name,
"schedule": schedule,
}
if params is not None:
body["params"] = params
if actions is not None:
body["actions"] = actions
if tags is not None:
body["tags"] = tags
if notify_when is not None:
body["notify_when"] = notify_when
if throttle is not None:
body["throttle"] = throttle
if alert_delay is not None:
body["alert_delay"] = alert_delay
if flapping is not None:
body["flapping"] = flapping
if artifacts is not None:
body["artifacts"] = artifacts
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path(f"/api/alerting/rule/{_quote(id)}", space_id)
return await self.perform_request("PUT", path, body=body)
[docs]
async def delete(
self,
*,
id: str,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Delete a rule.
Calls ``DELETE /api/alerting/rule/{id}``.
Args:
id: Rule ID.
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
Empty response on success (HTTP 204).
Raises:
ValueError: If ``id`` is missing.
NotFoundError: If the rule does not exist.
Example:
>>> await client.alerting.rule.delete(id="my-rule-id")
"""
if not id:
raise ValueError("Parameter 'id' is required")
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path(f"/api/alerting/rule/{_quote(id)}", space_id)
return await self.perform_request("DELETE", path)
[docs]
async def find(
self,
*,
search: str | None = None,
page: int | None = None,
per_page: int | None = None,
sort_field: str | None = None,
sort_order: str | None = None,
fields: list[str] | None = None,
filter: str | None = None,
default_search_operator: str | None = None,
search_fields: list[str] | None = None,
has_reference: dict[str, str] | None = None,
filter_consumers: list[str] | None = None,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Find rules matching the given criteria.
Calls ``GET /api/alerting/rules/_find``. Only parameters explicitly
provided are sent; server defaults apply otherwise (``page=1``,
``per_page=10``). Note that Kibana rejects ``sort_order`` when
``sort_field`` is not set.
Args:
search: Search string applied to ``search_fields``.
page: Page number (server default 1).
per_page: Items per page (server default 10).
sort_field: Rule attribute to sort by (e.g. ``name``).
sort_order: Sort order, ``asc`` or ``desc`` (requires
``sort_field``).
fields: Rule attributes to return (repeated query key).
filter: KQL filter string, e.g.
``alert.attributes.tags:production``.
default_search_operator: Operator for the search query,
``OR`` (default) or ``AND``.
search_fields: Rule attributes the ``search`` applies to
(repeated query key).
has_reference: Filter to rules referencing a saved object,
e.g. ``{"type": "tag", "id": "tag-1"}`` (sent as a JSON
string).
filter_consumers: List of consumers to filter by (sent as a
JSON array string).
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
Paginated rules matching the criteria (``total``, ``data``, ...).
Example:
>>> found = await client.alerting.rule.find(
... search="cpu", sort_field="name", sort_order="asc"
... )
>>> found["total"]
"""
params: dict[str, Any] = {}
if search is not None:
params["search"] = search
if page is not None:
params["page"] = page
if per_page is not None:
params["per_page"] = per_page
if sort_field is not None:
params["sort_field"] = sort_field
if sort_order is not None:
params["sort_order"] = sort_order
if fields is not None:
params["fields"] = fields
if filter is not None:
params["filter"] = filter
if default_search_operator is not None:
params["default_search_operator"] = default_search_operator
if search_fields is not None:
params["search_fields"] = search_fields
if has_reference is not None:
params["has_reference"] = has_reference
if filter_consumers is not None:
# Live Kibana 9.4.3 rejects a single repeated-key value for this
# param ("could not parse array value from json input"); it
# accepts a JSON array string for any number of values.
params["filter_consumers"] = json.dumps(filter_consumers)
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path("/api/alerting/rules/_find", space_id)
return await self.perform_request("GET", path, params=params or None)
[docs]
async def enable(
self,
*,
id: str,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Enable a rule.
Calls ``POST /api/alerting/rule/{id}/_enable``.
Args:
id: Rule ID.
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
Empty response on success (HTTP 204).
Raises:
ValueError: If ``id`` is missing.
NotFoundError: If the rule does not exist.
Example:
>>> await client.alerting.rule.enable(id="my-rule-id")
"""
if not id:
raise ValueError("Parameter 'id' is required")
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path(
f"/api/alerting/rule/{_quote(id)}/_enable", space_id
)
return await self.perform_request("POST", path)
[docs]
async def disable(
self,
*,
id: str,
untrack: bool | None = None,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Disable a rule.
Calls ``POST /api/alerting/rule/{id}/_disable``.
Args:
id: Rule ID.
untrack: Whether the rule's alerts should be untracked when the
rule is disabled.
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
Empty response on success (HTTP 204).
Raises:
ValueError: If ``id`` is missing.
NotFoundError: If the rule does not exist.
Example:
>>> await client.alerting.rule.disable(id="my-rule-id", untrack=True)
"""
if not id:
raise ValueError("Parameter 'id' is required")
body: dict[str, Any] | None = None
if untrack is not None:
body = {"untrack": untrack}
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path(
f"/api/alerting/rule/{_quote(id)}/_disable", space_id
)
return await self.perform_request("POST", path, body=body)
[docs]
async def mute_all(
self,
*,
id: str,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Mute all alerts for a rule.
Calls ``POST /api/alerting/rule/{id}/_mute_all``.
Args:
id: Rule ID.
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
Empty response on success (HTTP 204).
Raises:
ValueError: If ``id`` is missing.
NotFoundError: If the rule does not exist.
Example:
>>> await client.alerting.rule.mute_all(id="my-rule-id")
"""
if not id:
raise ValueError("Parameter 'id' is required")
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path(
f"/api/alerting/rule/{_quote(id)}/_mute_all", space_id
)
return await self.perform_request("POST", path)
[docs]
async def unmute_all(
self,
*,
id: str,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Unmute all alerts for a rule.
Calls ``POST /api/alerting/rule/{id}/_unmute_all``.
Args:
id: Rule ID.
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
Empty response on success (HTTP 204).
Raises:
ValueError: If ``id`` is missing.
NotFoundError: If the rule does not exist.
Example:
>>> await client.alerting.rule.unmute_all(id="my-rule-id")
"""
if not id:
raise ValueError("Parameter 'id' is required")
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path(
f"/api/alerting/rule/{_quote(id)}/_unmute_all", space_id
)
return await self.perform_request("POST", path)
[docs]
async def update_api_key(
self,
*,
id: str,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Update the API key for a rule.
Calls ``POST /api/alerting/rule/{id}/_update_api_key``. The rule's
API key is regenerated using the credentials of the caller.
Args:
id: Rule ID.
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
Empty response on success (HTTP 204).
Raises:
ValueError: If ``id`` is missing.
NotFoundError: If the rule does not exist.
ConflictError: If the rule is being updated concurrently.
Example:
>>> await client.alerting.rule.update_api_key(id="my-rule-id")
"""
if not id:
raise ValueError("Parameter 'id' is required")
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path(
f"/api/alerting/rule/{_quote(id)}/_update_api_key",
space_id,
)
return await self.perform_request("POST", path)
[docs]
async def snooze(
self,
*,
id: str,
schedule: dict[str, Any],
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Schedule a snooze for a rule.
Calls ``POST /api/alerting/rule/{id}/snooze_schedule``.
Generally available; added in 8.19.0.
Args:
id: Rule ID.
schedule: Snooze schedule object, e.g.
``{"custom": {"duration": "1h",
"start": "2026-01-01T00:00:00.000Z"}}``.
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
The created snooze schedule (includes the schedule ``id`` needed
to delete it via :meth:`unsnooze`).
Raises:
ValueError: If a required parameter is missing.
NotFoundError: If the rule does not exist.
Example:
>>> snoozed = await client.alerting.rule.snooze(
... id="my-rule-id",
... schedule={"custom": {"duration": "1h",
... "start": "2026-01-01T00:00:00.000Z"}},
... )
>>> schedule_id = snoozed["schedule"]["id"]
"""
if not id:
raise ValueError("Parameter 'id' is required")
if schedule is None:
raise ValueError("Parameter 'schedule' is required")
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path(
f"/api/alerting/rule/{_quote(id)}/snooze_schedule",
space_id,
)
return await self.perform_request("POST", path, body={"schedule": schedule})
[docs]
async def unsnooze(
self,
*,
rule_id: str,
schedule_id: str,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Delete a snooze schedule for a rule.
Calls ``DELETE /api/alerting/rule/{ruleId}/snooze_schedule/{scheduleId}``.
Generally available; added in 8.19.0.
Args:
rule_id: Rule ID.
schedule_id: Snooze schedule ID (returned by :meth:`snooze`).
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
Empty response on success (HTTP 204).
Raises:
ValueError: If a required parameter is missing.
NotFoundError: If the rule or schedule does not exist.
Example:
>>> await client.alerting.rule.unsnooze(
... rule_id="my-rule-id", schedule_id="schedule-uuid"
... )
"""
if not rule_id:
raise ValueError("Parameter 'rule_id' is required")
if not schedule_id:
raise ValueError("Parameter 'schedule_id' is required")
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path(
f"/api/alerting/rule/{_quote(rule_id)}/snooze_schedule/"
f"{_quote(schedule_id)}",
space_id,
)
return await self.perform_request("DELETE", path)
[docs]
async def mute_alert(
self,
*,
rule_id: str,
alert_id: str,
validate_alerts_existence: bool | None = None,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Mute a specific alert of a rule.
Calls ``POST /api/alerting/rule/{rule_id}/alert/{alert_id}/_mute``.
Args:
rule_id: Rule ID.
alert_id: Alert (instance) ID to mute.
validate_alerts_existence: Whether to validate that the alert
instance exists before muting (live 9.4.3 default: ``true``,
returning 404 for unknown alerts; pass ``False`` to mute
an alert that has not fired yet).
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
Empty response on success (HTTP 204).
Raises:
ValueError: If a required parameter is missing.
NotFoundError: If the rule or the alert instance does not exist.
Example:
>>> await client.alerting.rule.mute_alert(
... rule_id="my-rule-id", alert_id="server-1"
... )
"""
if not rule_id:
raise ValueError("Parameter 'rule_id' is required")
if not alert_id:
raise ValueError("Parameter 'alert_id' is required")
params: dict[str, Any] = {}
if validate_alerts_existence is not None:
params["validate_alerts_existence"] = validate_alerts_existence
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path(
f"/api/alerting/rule/{_quote(rule_id)}/alert/{_quote(alert_id)}/_mute",
space_id,
)
return await self.perform_request("POST", path, params=params or None)
[docs]
async def unmute_alert(
self,
*,
rule_id: str,
alert_id: str,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Unmute a specific alert of a rule.
Calls ``POST /api/alerting/rule/{rule_id}/alert/{alert_id}/_unmute``.
Args:
rule_id: Rule ID.
alert_id: Alert (instance) ID to unmute.
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
Empty response on success (HTTP 204).
Raises:
ValueError: If a required parameter is missing.
NotFoundError: If the rule does not exist.
Example:
>>> await client.alerting.rule.unmute_alert(
... rule_id="my-rule-id", alert_id="server-1"
... )
"""
if not rule_id:
raise ValueError("Parameter 'rule_id' is required")
if not alert_id:
raise ValueError("Parameter 'alert_id' is required")
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path(
f"/api/alerting/rule/{_quote(rule_id)}/alert/{_quote(alert_id)}/_unmute",
space_id,
)
return await self.perform_request("POST", path)
[docs]
class AsyncBackfillClient(AsyncNamespaceClient):
"""Async client for Kibana Alerting backfill operations.
Backfills run a rule over a historical time range. Only certain rule
types support backfills (e.g. detection rules); scheduling a backfill
for an unsupported rule type returns a per-item error in the response
body.
Example:
>>> await client.alerting.backfill.schedule(
... backfills=[{
... "rule_id": "my-rule-id",
... "ranges": [{"start": "2026-01-01T00:00:00.000Z",
... "end": "2026-01-01T12:00:00.000Z"}],
... }]
... )
"""
[docs]
async def schedule(
self,
*,
backfills: list[dict[str, Any]],
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Schedule backfill runs for one or more rules.
Calls ``POST /api/alerting/rules/backfill/_schedule``.
Args:
backfills: List of 1-100 backfill request objects. Each requires
``rule_id`` and ``ranges`` (list of ``{"start", "end"}`` ISO
8601 timestamps) and accepts an optional ``run_actions``
boolean.
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
A list with one result (or per-item error object) per requested
backfill.
Raises:
ValueError: If ``backfills`` is missing or empty.
BadRequestError: If the request payload is invalid.
Example:
>>> await client.alerting.backfill.schedule(
... backfills=[{
... "rule_id": "my-rule-id",
... "ranges": [{"start": "2026-01-01T00:00:00.000Z",
... "end": "2026-01-01T12:00:00.000Z"}],
... "run_actions": False,
... }]
... )
"""
if not backfills:
raise ValueError("Parameter 'backfills' is required")
body: Any = backfills
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path(
"/api/alerting/rules/backfill/_schedule", space_id
)
return await self.perform_request("POST", path, body=body)
[docs]
async def find(
self,
*,
rule_ids: str | None = None,
start: str | None = None,
end: str | None = None,
page: int | None = None,
per_page: int | None = None,
sort_field: str | None = None,
sort_order: str | None = None,
initiator: str | None = None,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Find backfills for rules.
Calls ``POST /api/alerting/rules/backfill/_find``. All filters are
query parameters; there is no request body.
Args:
rule_ids: Comma-separated list of rule IDs to filter by.
start: Backfills with a start time after this ISO 8601 timestamp.
end: Backfills with an end time before this ISO 8601 timestamp.
page: Page number (server default 1).
per_page: Items per page (server default 10).
sort_field: Field to sort by (``createdAt`` or ``start``).
sort_order: Sort order, ``asc`` or ``desc``.
initiator: Filter by who started the backfill (``user`` or
``system``).
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
Paginated backfills matching the criteria.
Example:
>>> await client.alerting.backfill.find(rule_ids="my-rule-id")
"""
params: dict[str, Any] = {}
if rule_ids is not None:
params["rule_ids"] = rule_ids
if start is not None:
params["start"] = start
if end is not None:
params["end"] = end
if page is not None:
params["page"] = page
if per_page is not None:
params["per_page"] = per_page
if sort_field is not None:
params["sort_field"] = sort_field
if sort_order is not None:
params["sort_order"] = sort_order
if initiator is not None:
params["initiator"] = initiator
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path("/api/alerting/rules/backfill/_find", space_id)
return await self.perform_request("POST", path, params=params or None)
[docs]
async def get(
self,
*,
id: str,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Get a backfill by ID.
Calls ``GET /api/alerting/rules/backfill/{id}``.
Args:
id: Backfill ID.
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
The backfill object.
Raises:
ValueError: If ``id`` is missing.
NotFoundError: If the backfill does not exist.
Example:
>>> await client.alerting.backfill.get(id="backfill-id")
"""
if not id:
raise ValueError("Parameter 'id' is required")
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path(
f"/api/alerting/rules/backfill/{_quote(id)}", space_id
)
return await self.perform_request("GET", path)
[docs]
async def delete(
self,
*,
id: str,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Delete a backfill by ID.
Calls ``DELETE /api/alerting/rules/backfill/{id}``.
Args:
id: Backfill ID.
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
Empty response on success (HTTP 204).
Raises:
ValueError: If ``id`` is missing.
NotFoundError: If the backfill does not exist.
Example:
>>> await client.alerting.backfill.delete(id="backfill-id")
"""
if not id:
raise ValueError("Parameter 'id' is required")
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path(
f"/api/alerting/rules/backfill/{_quote(id)}", space_id
)
return await self.perform_request("DELETE", path)
[docs]
class AsyncAlertingClient(AsyncNamespaceClient):
"""Async client for Kibana Alerting API operations.
Exposes rule operations via :attr:`rule`, backfill operations via
:attr:`backfill`, and framework-level endpoints (:meth:`health`,
:meth:`rule_types`) directly.
Example:
>>> from kibana import AsyncKibana
>>> client = AsyncKibana("http://localhost:5601", api_key="...")
>>> (await client.alerting.health())["alerting_framework_health"]
>>> await client.alerting.rule.find(search="cpu")
"""
[docs]
def __init__(
self,
client: AsyncKibana,
default_space_id: str | None = None,
validate_spaces: bool = True,
) -> None:
"""Initialize the AsyncAlertingClient.
Args:
client: The parent Kibana client instance.
default_space_id: Optional default space ID for all operations.
validate_spaces: Whether to validate space existence.
"""
super().__init__(
client,
default_space_id=default_space_id,
validate_spaces=validate_spaces,
)
self._rule = AsyncRulesClient(
client,
default_space_id=default_space_id,
validate_spaces=validate_spaces,
)
self._backfill = AsyncBackfillClient(
client,
default_space_id=default_space_id,
validate_spaces=validate_spaces,
)
@property
def rule(self) -> AsyncRulesClient:
"""Access rule operations."""
return self._rule
@property
def backfill(self) -> AsyncBackfillClient:
"""Access backfill operations."""
return self._backfill
[docs]
async def health(
self,
*,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Get the alerting framework health.
Calls ``GET /api/alerting/_health``.
Args:
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
Health details: ``is_sufficiently_secure``,
``has_permanent_encryption_key`` and
``alerting_framework_health``.
Example:
>>> health = await client.alerting.health()
>>> health["alerting_framework_health"]["execution_health"]
"""
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path("/api/alerting/_health", space_id)
return await self.perform_request("GET", path)
[docs]
async def rule_types(
self,
*,
space_id: str | None = None,
validate_spaces: bool | None = None,
) -> ObjectApiResponse[Any]:
"""Get the rule types available in the current space.
Calls ``GET /api/alerting/rule_types``. The response body is a JSON
array of rule type descriptors.
Args:
space_id: Optional space ID.
validate_spaces: Override space validation for this call.
Returns:
List of rule type descriptors (``id``, ``name``,
``action_groups``, ``authorized_consumers``, ...).
Example:
>>> types = await client.alerting.rule_types()
>>> [t["id"] for t in types]
"""
await self._maybe_validate_space(space_id, validate_spaces)
path = self._build_space_path("/api/alerting/rule_types", space_id)
return await self.perform_request("GET", path)