SlosClient

Client for the Kibana SLOs (Service Level Objectives) API.

SLOs let you set measurable targets (availability, latency, …) for your services based on Elasticsearch data and track error budgets against those targets. The SLO APIs require an appropriate license (Platinum or trial).

All endpoints are space-scoped: the official API paths are rooted at /s/{spaceId}/api/observability/slos. Every method therefore accepts space_id (None targets the default space) and validate_spaces to override space-existence validation per call.

class kibana._sync.client.slos.SlosClient(client, default_space_id=None, validate_spaces=True)[source]

Bases: NamespaceClient

Client for the Kibana SLOs (Service Level Objectives) API.

SLOs let you set measurable targets (availability, latency, …) for your services based on Elasticsearch data and track error budgets against those targets. The SLO APIs require an appropriate license (Platinum or trial) and Kibana 8.12+ / 9.x.

All endpoints are space-scoped: the official API paths are rooted at /s/{spaceId}/api/observability/slos. Every method therefore accepts space_id (None targets the default space) and validate_spaces to override space-existence validation per call.

Example

>>> from kibana import Kibana
>>> client = Kibana("http://localhost:5601", api_key="...")
>>>
>>> # Create an SLO based on a custom KQL indicator
>>> slo = client.slos.create(
...     name="my-service availability",
...     description="99% of requests are good over 30 days",
...     indicator={
...         "type": "sli.kql.custom",
...         "params": {
...             "index": "my-service-logs",
...             "good": "status: ok",
...             "total": "",
...             "timestampField": "@timestamp",
...         },
...     },
...     time_window={"duration": "30d", "type": "rolling"},
...     budgeting_method="occurrences",
...     objective={"target": 0.99},
... )
>>> slo_id = slo.body["id"]
>>>
>>> # Fetch it back, then clean up
>>> client.slos.get(slo_id=slo_id)
>>> client.slos.delete(slo_id=slo_id)

Creating SLOs

Create an SLO with the create() method:

from kibana import Kibana

client = Kibana("http://localhost:5601", api_key="your_api_key")

# Create an SLO based on a custom KQL indicator
slo = client.slos.create(
    name="my-service availability",
    description="99% of requests are good over 30 days",
    indicator={
        "type": "sli.kql.custom",
        "params": {
            "index": "my-service-logs",
            "good": "status: ok",
            "total": "",
            "timestampField": "@timestamp",
        },
    },
    time_window={"duration": "30d", "type": "rolling"},
    budgeting_method="occurrences",
    objective={"target": 0.99},
)

slo_id = slo.body["id"]

Finding and Retrieving SLOs

# Get a single SLO (with its computed summary)
slo = client.slos.get(slo_id=slo_id)

# Find SLOs with a KQL query
results = client.slos.find(kql_query="slo.name:my-service*")
for slo in results.body["results"]:
    print(slo["name"], slo["summary"]["status"])

# List SLO definitions
definitions = client.slos.find_definitions(search="my-service*")

SLO Lifecycle

# Disable and re-enable an SLO
client.slos.disable(slo_id=slo_id)
client.slos.enable(slo_id=slo_id)

# Reset an SLO (recompute rollup data from scratch)
client.slos.reset(slo_id=slo_id)

# Delete an SLO
client.slos.delete(slo_id=slo_id)

Bulk Operations

# Bulk delete SLOs (asynchronous server-side task)
task = client.slos.bulk_delete(slo_ids=[slo_id, "another-slo-id"])

# Check the status of the bulk delete task
status = client.slos.bulk_delete_status(task_id=task.body["taskId"])
find(*, kql_query=None, size=None, search_after=None, page=None, per_page=None, sort_by=None, sort_direction=None, hide_stale=None, space_id=None, validate_spaces=None)[source]

Get a paginated list of SLOs with their summaries.

GET /api/observability/slos

Parameters:
  • kql_query (str | None) – A valid KQL query to filter SLOs with, e.g. 'slo.name:latency* and slo.tags : "prod"'.

  • size (int | None) – The page size to use for cursor-based pagination (must be used together with search_after). Default 1.

  • search_after (list[str] | None) – The cursor to use for fetching the results from, when using a cursor-based pagination.

  • page (int | None) – The page number to return (default 1). Offset-based pagination; mutually exclusive with size/search_after.

  • per_page (int | None) – The number of SLOs to return per page (default 25, maximum 5000).

  • sort_by (str | None) – Sort by field. One of "sli_value", "status", "error_budget_consumed", "error_budget_remaining" (default "status").

  • sort_direction (str | None) – Sort order, "asc" (default) or "desc".

  • hide_stale (bool | None) – Hide stale SLOs from the list as defined by the stale SLO threshold in SLO settings.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with page, perPage, total and results (a list of SLOs with their summary).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> resp = client.slos.find(kql_query="slo.name:my-service*")
>>> for slo in resp.body["results"]:
...     print(slo["name"], slo["summary"]["status"])
create(*, name, description, indicator, time_window, budgeting_method, objective, id=None, settings=None, group_by=None, tags=None, artifacts=None, space_id=None, validate_spaces=None)[source]

Create an SLO.

POST /api/observability/slos

Parameters:
  • name (str) – A name for the SLO.

  • description (str) – A description for the SLO.

  • indicator (dict[str, Any]) – The indicator (SLI) definition, e.g. {"type": "sli.kql.custom", "params": {...}}. Supported types: sli.apm.transactionDuration, sli.apm.transactionErrorRate, sli.kql.custom, sli.metric.custom, sli.histogram.custom, sli.metric.timeslice.

  • time_window (dict[str, Any]) – The SLO time window, e.g. {"duration": "30d", "type": "rolling"} or {"duration": "1M", "type": "calendarAligned"}.

  • budgeting_method (str) – "occurrences" or "timeslices".

  • objective (dict[str, Any]) – The objective, e.g. {"target": 0.99}; timeslices budgeting also uses timesliceTarget and timesliceWindow.

  • id (str | None) – An optional unique identifier for the SLO (8-36 characters); autogenerated when omitted.

  • settings (dict[str, Any] | None) – Optional settings such as syncDelay, frequency, syncField and preventInitialBackfill.

  • group_by (str | list[str] | None) – Optional field or list of fields to generate one SLO per distinct value (e.g. "service.name").

  • tags (list[str] | None) – Optional list of tags.

  • artifacts (dict[str, Any] | None) – Optional links to related assets, e.g. {"dashboards": [{"id": "..."}]}.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse containing the id of the created SLO.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> resp = client.slos.create(
...     name="availability",
...     description="99% good over 7 days",
...     indicator={
...         "type": "sli.kql.custom",
...         "params": {
...             "index": "logs-*",
...             "good": "status: ok",
...             "total": "",
...             "timestampField": "@timestamp",
...         },
...     },
...     time_window={"duration": "7d", "type": "rolling"},
...     budgeting_method="occurrences",
...     objective={"target": 0.99},
... )
>>> print(resp.body["id"])
get(*, slo_id, instance_id=None, space_id=None, validate_spaces=None)[source]

Get an SLO by its identifier.

GET /api/observability/slos/{sloId}

Parameters:
  • slo_id (str) – The SLO identifier.

  • instance_id (str | None) – The SLO instance identifier when retrieving a specific instance of a grouped SLO.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with the SLO definition and its summary.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> slo = client.slos.get(slo_id="my-slo-id")
>>> print(slo.body["name"], slo.body["enabled"])
update(*, slo_id, name=None, description=None, indicator=None, time_window=None, budgeting_method=None, objective=None, settings=None, group_by=None, tags=None, artifacts=None, space_id=None, validate_spaces=None)[source]

Update an SLO.

PUT /api/observability/slos/{sloId}

Performs a partial update: only the provided fields are changed. Updating fields that are part of the rollup transform (indicator, time window, budgeting method, objective, group_by or settings) triggers a re-computation of the SLO data.

Parameters:
  • slo_id (str) – The SLO identifier.

  • name (str | None) – A new name for the SLO.

  • description (str | None) – A new description for the SLO.

  • indicator (dict[str, Any] | None) – The indicator (SLI) definition.

  • time_window (dict[str, Any] | None) – The SLO time window.

  • budgeting_method (str | None) – "occurrences" or "timeslices".

  • objective (dict[str, Any] | None) – The objective, e.g. {"target": 0.99}.

  • settings (dict[str, Any] | None) – Settings such as syncDelay, frequency, syncField and preventInitialBackfill.

  • group_by (str | list[str] | None) – Field or list of fields to generate one SLO per distinct value.

  • tags (list[str] | None) – List of tags.

  • artifacts (dict[str, Any] | None) – Links to related assets, e.g. {"dashboards": [{"id": "..."}]}.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with the updated SLO definition.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> resp = client.slos.update(
...     slo_id="my-slo-id",
...     description="Updated description",
...     tags=["prod"],
... )
delete(*, slo_id, space_id=None, validate_spaces=None)[source]

Delete an SLO and its associated summary and rollup data.

DELETE /api/observability/slos/{sloId}

Parameters:
  • slo_id (str) – The SLO identifier.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with an empty body (HTTP 204 on success).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> client.slos.delete(slo_id="my-slo-id")
enable(*, slo_id, space_id=None, validate_spaces=None)[source]

Enable an SLO, resuming data rollup and summary computation.

POST /api/observability/slos/{sloId}/enable

Parameters:
  • slo_id (str) – The SLO identifier.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with an empty body (HTTP 204 on success).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> client.slos.enable(slo_id="my-slo-id")
disable(*, slo_id, space_id=None, validate_spaces=None)[source]

Disable an SLO, stopping data rollup and summary computation.

POST /api/observability/slos/{sloId}/disable

Parameters:
  • slo_id (str) – The SLO identifier.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with an empty body (HTTP 204 on success).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> client.slos.disable(slo_id="my-slo-id")
reset(*, slo_id, space_id=None, validate_spaces=None)[source]

Reset an SLO, deleting its data and recomputing it from scratch.

POST /api/observability/slos/{sloId}/_reset

Parameters:
  • slo_id (str) – The SLO identifier.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with the reset SLO definition.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> resp = client.slos.reset(slo_id="my-slo-id")
>>> print(resp.body["version"])
delete_instances(*, instances, space_id=None, validate_spaces=None)[source]

Batch delete rollup and summary data for SLO instances.

POST /api/observability/slos/_delete_instances

Deletes the rollup and summary data for the given list of SLO id / instance id pairs. Useful for removing stale data of instances of a grouped SLO that no longer receive updates.

Parameters:
  • instances (list[dict[str, Any]]) – A list of {"sloId": ..., "instanceId": ...} objects identifying the instances to delete.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with an empty body (HTTP 204 on success).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> client.slos.delete_instances(
...     instances=[
...         {"sloId": "my-slo-id", "instanceId": "my-service"},
...     ]
... )
bulk_delete(*, slo_ids, space_id=None, validate_spaces=None)[source]

Bulk delete SLO definitions and their summary and rollup data.

POST /api/observability/slos/_bulk_delete

The deletion occurs asynchronously: the response contains a taskId that can be polled with bulk_delete_status() to retrieve the operation outcome.

Parameters:
  • slo_ids (list[str]) – A list of SLO definition ids to delete.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse containing the taskId of the bulk deletion.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> resp = client.slos.bulk_delete(slo_ids=["slo-1", "slo-2"])
>>> status = client.slos.bulk_delete_status(
...     task_id=resp.body["taskId"]
... )
bulk_delete_status(*, task_id, space_id=None, validate_spaces=None)[source]

Retrieve the status of an asynchronous bulk deletion.

GET /api/observability/slos/_bulk_delete/{taskId}

Parameters:
  • task_id (str) – The task id returned by bulk_delete().

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with isDone, per-SLO results and an optional error.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> status = client.slos.bulk_delete_status(task_id="...")
>>> if status.body["isDone"]:
...     for result in status.body["results"]:
...         print(result["id"], result["success"])
bulk_purge_rollup(*, slo_ids, purge_policy, space_id=None, validate_spaces=None)[source]

Batch delete rollup and summary data according to a purge policy.

POST /api/observability/slos/_bulk_purge_rollup

The deletion occurs asynchronously (Elasticsearch delete-by-query) and the response contains an Elasticsearch task id.

Note

Kibana 9.4.3 expects purgeType values "fixed_age" / "fixed_time" (snake_case), although the official OpenAPI spec documents "fixed-age" / "fixed-time".

Parameters:
  • slo_ids (list[str]) – A list of SLO ids whose rollup data should be purged.

  • purge_policy (dict[str, Any]) – Policy dictating which SLI documents to purge, either {"purgeType": "fixed_age", "age": "7d"} or {"purgeType": "fixed_time", "timestamp": "2024-01-01T00:00:00.000Z"}.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse containing the Elasticsearch taskId of the purge operation.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> resp = client.slos.bulk_purge_rollup(
...     slo_ids=["my-slo-id"],
...     purge_policy={"purgeType": "fixed_age", "age": "30d"},
... )
>>> print(resp.body["taskId"])
find_definitions(*, include_outdated_only=None, include_health=None, tags=None, search=None, page=None, per_page=None, space_id=None, validate_spaces=None)[source]

Get the SLO definitions (without computed summaries).

GET /api/observability/slos/_definitions

Note

The official OpenAPI spec documents this operation under /internal/observability/slos/_definitions, but Kibana 9.4.3 serves it on the public /api/observability/slos/_definitions path (the internal path returns 404); this client uses the public path.

Parameters:
  • include_outdated_only (bool | None) – Indicates if the API returns only outdated SLOs or all SLOs.

  • include_health (bool | None) – Include the health of the SLO transforms in the response.

  • tags (str | None) – Filter SLOs by tag (comma-separated list of tags).

  • search (str | None) – Filter SLOs by name, e.g. "my-service*".

  • page (int | None) – The page number to return.

  • per_page (int | None) – The number of SLOs to return per page (default 100, maximum 1000).

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with page, perPage, total and results (a list of SLO definitions).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> resp = client.slos.find_definitions(search="my-service*")
>>> print(resp.body["total"])
__init__(client, default_space_id=None, validate_spaces=True)

Initialize NamespaceClient with optional space support.

Parameters:
  • client (BaseClient) – Parent BaseClient instance to delegate requests to

  • default_space_id (str | None) – Optional default space ID for all operations

  • validate_spaces (bool) – Whether to validate space existence (default: True)

perform_request(method, path, *, params=None, headers=None, body=None)

Perform an HTTP request via the parent client with space context enhancement.

Parameters:
  • method (str) – HTTP method (GET, POST, PUT, DELETE, etc.)

  • path (str) – API endpoint path

  • params (dict[str, Any] | None) – Query parameters

  • headers (dict[str, str] | None) – Request headers

  • body (dict[str, Any] | None) – Request body

Returns:

API response

Raises:

ApiError – If the API returns an error response (enhanced with space context)

Return type:

ObjectApiResponse[Any]

AsyncSlosClient

Asynchronous version of the SlosClient for use with async/await syntax.

class kibana._async.client.slos.AsyncSlosClient(client, default_space_id=None, validate_spaces=True)[source]

Bases: AsyncNamespaceClient

Async client for the Kibana SLOs (Service Level Objectives) API.

SLOs let you set measurable targets (availability, latency, …) for your services based on Elasticsearch data and track error budgets against those targets. The SLO APIs require an appropriate license (Platinum or trial) and Kibana 8.12+ / 9.x.

All endpoints are space-scoped: the official API paths are rooted at /s/{spaceId}/api/observability/slos. Every method therefore accepts space_id (None targets the default space) and validate_spaces to override space-existence validation per call.

Example

>>> from kibana import AsyncKibana
>>> client = AsyncKibana("http://localhost:5601", api_key="...")
>>>
>>> # Create an SLO based on a custom KQL indicator
>>> slo = await client.slos.create(
...     name="my-service availability",
...     description="99% of requests are good over 30 days",
...     indicator={
...         "type": "sli.kql.custom",
...         "params": {
...             "index": "my-service-logs",
...             "good": "status: ok",
...             "total": "",
...             "timestampField": "@timestamp",
...         },
...     },
...     time_window={"duration": "30d", "type": "rolling"},
...     budgeting_method="occurrences",
...     objective={"target": 0.99},
... )
>>> slo_id = slo.body["id"]
>>>
>>> # Fetch it back, then clean up
>>> await client.slos.get(slo_id=slo_id)
>>> await client.slos.delete(slo_id=slo_id)

Usage

The AsyncSlosClient provides the same methods as SlosClient but all methods are async and must be awaited:

from kibana import AsyncKibana
import asyncio

async def main():
    async with AsyncKibana("http://localhost:5601") as client:
        # Create an SLO (async)
        slo = await client.slos.create(
            name="async availability",
            description="Availability SLO created asynchronously",
            indicator={
                "type": "sli.kql.custom",
                "params": {
                    "index": "my-service-logs",
                    "good": "status: ok",
                    "total": "",
                    "timestampField": "@timestamp",
                },
            },
            time_window={"duration": "30d", "type": "rolling"},
            budgeting_method="occurrences",
            objective={"target": 0.99},
        )

        # Find SLOs (async)
        results = await client.slos.find()

        # Delete (async)
        await client.slos.delete(slo_id=slo.body["id"])

asyncio.run(main())
async find(*, kql_query=None, size=None, search_after=None, page=None, per_page=None, sort_by=None, sort_direction=None, hide_stale=None, space_id=None, validate_spaces=None)[source]

Get a paginated list of SLOs with their summaries.

GET /api/observability/slos

Parameters:
  • kql_query (str | None) – A valid KQL query to filter SLOs with, e.g. 'slo.name:latency* and slo.tags : "prod"'.

  • size (int | None) – The page size to use for cursor-based pagination (must be used together with search_after). Default 1.

  • search_after (list[str] | None) – The cursor to use for fetching the results from, when using a cursor-based pagination.

  • page (int | None) – The page number to return (default 1). Offset-based pagination; mutually exclusive with size/search_after.

  • per_page (int | None) – The number of SLOs to return per page (default 25, maximum 5000).

  • sort_by (str | None) – Sort by field. One of "sli_value", "status", "error_budget_consumed", "error_budget_remaining" (default "status").

  • sort_direction (str | None) – Sort order, "asc" (default) or "desc".

  • hide_stale (bool | None) – Hide stale SLOs from the list as defined by the stale SLO threshold in SLO settings.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with page, perPage, total and results (a list of SLOs with their summary).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> resp = await client.slos.find(kql_query="slo.name:my-service*")
>>> for slo in resp.body["results"]:
...     print(slo["name"], slo["summary"]["status"])
async create(*, name, description, indicator, time_window, budgeting_method, objective, id=None, settings=None, group_by=None, tags=None, artifacts=None, space_id=None, validate_spaces=None)[source]

Create an SLO.

POST /api/observability/slos

Parameters:
  • name (str) – A name for the SLO.

  • description (str) – A description for the SLO.

  • indicator (dict[str, Any]) – The indicator (SLI) definition, e.g. {"type": "sli.kql.custom", "params": {...}}. Supported types: sli.apm.transactionDuration, sli.apm.transactionErrorRate, sli.kql.custom, sli.metric.custom, sli.histogram.custom, sli.metric.timeslice.

  • time_window (dict[str, Any]) – The SLO time window, e.g. {"duration": "30d", "type": "rolling"} or {"duration": "1M", "type": "calendarAligned"}.

  • budgeting_method (str) – "occurrences" or "timeslices".

  • objective (dict[str, Any]) – The objective, e.g. {"target": 0.99}; timeslices budgeting also uses timesliceTarget and timesliceWindow.

  • id (str | None) – An optional unique identifier for the SLO (8-36 characters); autogenerated when omitted.

  • settings (dict[str, Any] | None) – Optional settings such as syncDelay, frequency, syncField and preventInitialBackfill.

  • group_by (str | list[str] | None) – Optional field or list of fields to generate one SLO per distinct value (e.g. "service.name").

  • tags (list[str] | None) – Optional list of tags.

  • artifacts (dict[str, Any] | None) – Optional links to related assets, e.g. {"dashboards": [{"id": "..."}]}.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse containing the id of the created SLO.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> resp = await client.slos.create(
...     name="availability",
...     description="99% good over 7 days",
...     indicator={
...         "type": "sli.kql.custom",
...         "params": {
...             "index": "logs-*",
...             "good": "status: ok",
...             "total": "",
...             "timestampField": "@timestamp",
...         },
...     },
...     time_window={"duration": "7d", "type": "rolling"},
...     budgeting_method="occurrences",
...     objective={"target": 0.99},
... )
>>> print(resp.body["id"])
async get(*, slo_id, instance_id=None, space_id=None, validate_spaces=None)[source]

Get an SLO by its identifier.

GET /api/observability/slos/{sloId}

Parameters:
  • slo_id (str) – The SLO identifier.

  • instance_id (str | None) – The SLO instance identifier when retrieving a specific instance of a grouped SLO.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with the SLO definition and its summary.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> slo = await client.slos.get(slo_id="my-slo-id")
>>> print(slo.body["name"], slo.body["enabled"])
async update(*, slo_id, name=None, description=None, indicator=None, time_window=None, budgeting_method=None, objective=None, settings=None, group_by=None, tags=None, artifacts=None, space_id=None, validate_spaces=None)[source]

Update an SLO.

PUT /api/observability/slos/{sloId}

Performs a partial update: only the provided fields are changed. Updating fields that are part of the rollup transform (indicator, time window, budgeting method, objective, group_by or settings) triggers a re-computation of the SLO data.

Parameters:
  • slo_id (str) – The SLO identifier.

  • name (str | None) – A new name for the SLO.

  • description (str | None) – A new description for the SLO.

  • indicator (dict[str, Any] | None) – The indicator (SLI) definition.

  • time_window (dict[str, Any] | None) – The SLO time window.

  • budgeting_method (str | None) – "occurrences" or "timeslices".

  • objective (dict[str, Any] | None) – The objective, e.g. {"target": 0.99}.

  • settings (dict[str, Any] | None) – Settings such as syncDelay, frequency, syncField and preventInitialBackfill.

  • group_by (str | list[str] | None) – Field or list of fields to generate one SLO per distinct value.

  • tags (list[str] | None) – List of tags.

  • artifacts (dict[str, Any] | None) – Links to related assets, e.g. {"dashboards": [{"id": "..."}]}.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with the updated SLO definition.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> resp = await client.slos.update(
...     slo_id="my-slo-id",
...     description="Updated description",
...     tags=["prod"],
... )
async delete(*, slo_id, space_id=None, validate_spaces=None)[source]

Delete an SLO and its associated summary and rollup data.

DELETE /api/observability/slos/{sloId}

Parameters:
  • slo_id (str) – The SLO identifier.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with an empty body (HTTP 204 on success).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> await client.slos.delete(slo_id="my-slo-id")
async enable(*, slo_id, space_id=None, validate_spaces=None)[source]

Enable an SLO, resuming data rollup and summary computation.

POST /api/observability/slos/{sloId}/enable

Parameters:
  • slo_id (str) – The SLO identifier.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with an empty body (HTTP 204 on success).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> await client.slos.enable(slo_id="my-slo-id")
async disable(*, slo_id, space_id=None, validate_spaces=None)[source]

Disable an SLO, stopping data rollup and summary computation.

POST /api/observability/slos/{sloId}/disable

Parameters:
  • slo_id (str) – The SLO identifier.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with an empty body (HTTP 204 on success).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> await client.slos.disable(slo_id="my-slo-id")
async reset(*, slo_id, space_id=None, validate_spaces=None)[source]

Reset an SLO, deleting its data and recomputing it from scratch.

POST /api/observability/slos/{sloId}/_reset

Parameters:
  • slo_id (str) – The SLO identifier.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with the reset SLO definition.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> resp = await client.slos.reset(slo_id="my-slo-id")
>>> print(resp.body["version"])
async delete_instances(*, instances, space_id=None, validate_spaces=None)[source]

Batch delete rollup and summary data for SLO instances.

POST /api/observability/slos/_delete_instances

Deletes the rollup and summary data for the given list of SLO id / instance id pairs. Useful for removing stale data of instances of a grouped SLO that no longer receive updates.

Parameters:
  • instances (list[dict[str, Any]]) – A list of {"sloId": ..., "instanceId": ...} objects identifying the instances to delete.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with an empty body (HTTP 204 on success).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> await client.slos.delete_instances(
...     instances=[
...         {"sloId": "my-slo-id", "instanceId": "my-service"},
...     ]
... )
async bulk_delete(*, slo_ids, space_id=None, validate_spaces=None)[source]

Bulk delete SLO definitions and their summary and rollup data.

POST /api/observability/slos/_bulk_delete

The deletion occurs asynchronously: the response contains a taskId that can be polled with bulk_delete_status() to retrieve the operation outcome.

Parameters:
  • slo_ids (list[str]) – A list of SLO definition ids to delete.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse containing the taskId of the bulk deletion.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> resp = await client.slos.bulk_delete(slo_ids=["slo-1", "slo-2"])
>>> status = await client.slos.bulk_delete_status(
...     task_id=resp.body["taskId"]
... )
async bulk_delete_status(*, task_id, space_id=None, validate_spaces=None)[source]

Retrieve the status of an asynchronous bulk deletion.

GET /api/observability/slos/_bulk_delete/{taskId}

Parameters:
  • task_id (str) – The task id returned by bulk_delete().

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with isDone, per-SLO results and an optional error.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> status = await client.slos.bulk_delete_status(task_id="...")
>>> if status.body["isDone"]:
...     for result in status.body["results"]:
...         print(result["id"], result["success"])
async bulk_purge_rollup(*, slo_ids, purge_policy, space_id=None, validate_spaces=None)[source]

Batch delete rollup and summary data according to a purge policy.

POST /api/observability/slos/_bulk_purge_rollup

The deletion occurs asynchronously (Elasticsearch delete-by-query) and the response contains an Elasticsearch task id.

Note

Kibana 9.4.3 expects purgeType values "fixed_age" / "fixed_time" (snake_case), although the official OpenAPI spec documents "fixed-age" / "fixed-time".

Parameters:
  • slo_ids (list[str]) – A list of SLO ids whose rollup data should be purged.

  • purge_policy (dict[str, Any]) – Policy dictating which SLI documents to purge, either {"purgeType": "fixed_age", "age": "7d"} or {"purgeType": "fixed_time", "timestamp": "2024-01-01T00:00:00.000Z"}.

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse containing the Elasticsearch taskId of the purge operation.

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> resp = await client.slos.bulk_purge_rollup(
...     slo_ids=["my-slo-id"],
...     purge_policy={"purgeType": "fixed_age", "age": "30d"},
... )
>>> print(resp.body["taskId"])
async find_definitions(*, include_outdated_only=None, include_health=None, tags=None, search=None, page=None, per_page=None, space_id=None, validate_spaces=None)[source]

Get the SLO definitions (without computed summaries).

GET /api/observability/slos/_definitions

Note

The official OpenAPI spec documents this operation under /internal/observability/slos/_definitions, but Kibana 9.4.3 serves it on the public /api/observability/slos/_definitions path (the internal path returns 404); this client uses the public path.

Parameters:
  • include_outdated_only (bool | None) – Indicates if the API returns only outdated SLOs or all SLOs.

  • include_health (bool | None) – Include the health of the SLO transforms in the response.

  • tags (str | None) – Filter SLOs by tag (comma-separated list of tags).

  • search (str | None) – Filter SLOs by name, e.g. "my-service*".

  • page (int | None) – The page number to return.

  • per_page (int | None) – The number of SLOs to return per page (default 100, maximum 1000).

  • space_id (str | None) – Space to operate on; None targets the default space.

  • validate_spaces (bool | None) – Override space-existence validation for this call.

Returns:

ObjectApiResponse with page, perPage, total and results (a list of SLO definitions).

Raises:
Return type:

ObjectApiResponse[Any]

Example

>>> resp = await client.slos.find_definitions(search="my-service*")
>>> print(resp.body["total"])
__init__(client, default_space_id=None, validate_spaces=True)

Initialize AsyncNamespaceClient with optional space support.

Parameters:
  • client (AsyncBaseClient) – Parent AsyncBaseClient instance to delegate requests to

  • default_space_id (str | None) – Optional default space ID for all operations

  • validate_spaces (bool) – Whether to validate space existence (default: True)

async perform_request(method, path, *, params=None, headers=None, body=None)

Perform an async HTTP request via the parent client with space context enhancement.

Parameters:
  • method (str) – HTTP method (GET, POST, PUT, DELETE, etc.)

  • path (str) – API endpoint path

  • params (dict[str, Any] | None) – Query parameters

  • headers (dict[str, str] | None) – Request headers

  • body (dict[str, Any] | None) – Request body

Returns:

API response

Raises:

ApiError – If the API returns an error response (enhanced with space context)

Return type:

ObjectApiResponse[Any]