Exceptions¶

Exception classes for error handling in the Kibana Python client.

The kibana-py library provides a comprehensive exception hierarchy for handling various error conditions that may occur when interacting with the Kibana API.

Exception Hierarchy¶

All exceptions inherit from the base KibanaException class:

KibanaException
├── ApiError
│   ├── BadRequestError (400)
│   ├── AuthenticationException (401)
│   ├── AuthorizationException (403)
│   ├── NotFoundError (404)
│   └── ConflictError (409)
├── TransportError
│   ├── ConnectionError
│   │   ├── ConnectionTimeout
│   │   └── SSLError
│   └── SerializationError
└── SpaceError
    ├── SpaceNotFoundError
    └── InvalidSpaceIdError

Base Exceptions¶

KibanaException¶

class kibana.KibanaException[source]¶

Bases: Exception

Base exception for all Kibana client errors.

All exceptions raised by the Kibana client inherit from this class, making it easy to catch all Kibana-related errors.

Example

>>> from kibana import Kibana
>>> from kibana.exceptions import KibanaException
>>>
>>> client = Kibana("http://localhost:5601")
>>> try:
...     client.actions.get(id="nonexistent")
... except KibanaException as e:
...     print(f"Kibana error: {e}")

Base exception for all Kibana client errors. Catch this to handle any error from the Kibana client.

from kibana import Kibana
from kibana.exceptions import KibanaException

client = Kibana("http://localhost:5601")

try:
    # Perform operations
    connector = client.actions.create(
        name="Test",
        connector_type_id=".webhook",
        config={"url": "https://example.com"}
    )
except KibanaException as e:
    print(f"Kibana error occurred: {e}")

API Exceptions¶

ApiError¶

class kibana.ApiError(message, meta, body)[source]¶

Bases: KibanaException

API returned an error response.

Raised when the Kibana API returns an HTTP error status code (4xx or 5xx). Contains the error message, response metadata, and full response body for debugging.

message¶: Human-readable error message extracted from the response.

meta¶: API response metadata including status code, headers, and timing.

body¶: Full response body from the API (may contain additional error details).

status_code¶: HTTP status code of the error response.

Example

>>> try:
...     client.actions.create(name="", connector_type_id=".webhook", config={})
... except ApiError as e:
...     print(f"API Error: {e.message}")
...     print(f"Status Code: {e.status_code}")
...     print(f"Response Body: {e.body}")

Base class for API-level errors. Contains response metadata and body.

Attributes:

message (str): Error message
meta (ApiResponseMeta): Response metadata including status code
body (Any): Response body containing error details
status_code (int): HTTP status code

from kibana.exceptions import ApiError

try:
    result = client.actions.get(id="nonexistent")
except ApiError as e:
    print(f"API error: {e.message}")
    print(f"Status code: {e.status_code}")
    print(f"Response body: {e.body}")

__init__(message, meta, body)[source]¶

Initialize ApiError with error details.

Parameters:

message (str) – Human-readable error message.
meta (ApiResponseMeta) – API response metadata from elastic-transport.
body (Any) – Full response body from the API.

BadRequestError¶

class kibana.BadRequestError(message, meta, body)[source]¶

Bases: ApiError

Bad request (400).

Raised when the request is malformed or contains invalid parameters. This indicates a problem with the request data, such as: - Missing required parameters - Invalid parameter values - Malformed JSON - Invalid configuration

Example

>>> try:
...     client.actions.create(
...         name="",  # Empty name
...         connector_type_id=".webhook",
...         config={}
...     )
... except BadRequestError as e:
...     print(f"Invalid request: {e.message}")
...     print("Check the request parameters")

Raised when the API returns a 400 Bad Request status code.

This typically indicates invalid parameters or malformed requests.

from kibana.exceptions import BadRequestError

try:
    # Invalid connector configuration
    connector = client.actions.create(
        name="Test",
        connector_type_id=".webhook",
        config={}  # Missing required 'url' field
    )
except BadRequestError as e:
    print(f"Invalid request: {e.message}")
    print(f"Details: {e.body}")

AuthenticationException¶

class kibana.AuthenticationException(message, meta, body)[source]¶

Bases: ApiError

Authentication failed (401).

Raised when the provided credentials are invalid or missing. This indicates that the API key, basic auth credentials, or bearer token is incorrect or has expired.

Example

>>> client = Kibana("http://localhost:5601", api_key="invalid")
>>> try:
...     client.status.get_status()
... except AuthenticationException as e:
...     print(f"Authentication failed: {e.message}")
...     print("Please check your API key or credentials")

Raised when authentication fails (401 Unauthorized).

This indicates invalid or missing authentication credentials.

from kibana.exceptions import AuthenticationException

try:
    client = Kibana(
        "http://localhost:5601",
        api_key="invalid_key"
    )
    status = client.status.get_status()
except AuthenticationException as e:
    print(f"Authentication failed: {e.message}")
    print("Please check your API key or credentials")

AuthorizationException¶

class kibana.AuthorizationException(message, meta, body)[source]¶

Bases: ApiError

Authorization failed (403).

Raised when the authenticated user does not have sufficient privileges to perform the requested operation. The credentials are valid, but the user lacks the necessary permissions.

Example

>>> try:
...     client.spaces.delete(id="default")  # Cannot delete default space
... except AuthorizationException as e:
...     print(f"Permission denied: {e.message}")
...     print("User lacks required privileges")

Raised when authorization fails (403 Forbidden).

This indicates the authenticated user lacks permissions for the requested operation.

from kibana.exceptions import AuthorizationException

try:
    # User doesn't have permission to create connectors
    connector = client.actions.create(
        name="Test",
        connector_type_id=".webhook",
        config={"url": "https://example.com"}
    )
except AuthorizationException as e:
    print(f"Permission denied: {e.message}")
    print("User lacks required privileges")

NotFoundError¶

class kibana.NotFoundError(message, meta, body)[source]¶

Bases: ApiError

Resource not found (404).

Raised when the requested resource (connector, space, saved object, etc.) does not exist.

Example

>>> try:
...     client.actions.get(id="nonexistent-connector")
... except NotFoundError as e:
...     print(f"Resource not found: {e.message}")
...     print("The connector may have been deleted")

Raised when a resource is not found (404 Not Found).

from kibana.exceptions import NotFoundError

try:
    connector = client.actions.get(id="nonexistent-id")
except NotFoundError as e:
    print(f"Resource not found: {e.message}")

try:
    space = client.spaces.get(id="nonexistent-space")
except NotFoundError as e:
    print(f"Space not found: {e.message}")

ConflictError¶

class kibana.ConflictError(message, meta, body)[source]¶

Bases: ApiError

Conflict error (409).

Raised when the operation conflicts with the current state of the resource. Common causes include: - Creating a resource with an ID that already exists - Updating a resource with an outdated version - Deleting a resource that has dependencies

Example

>>> try:
...     client.spaces.create(id="marketing", name="Marketing")
...     client.spaces.create(id="marketing", name="Marketing 2")  # Duplicate
... except ConflictError as e:
...     print(f"Conflict: {e.message}")
...     print("A space with this ID already exists")

Raised when a conflict occurs (409 Conflict).

This typically indicates a resource already exists or a version conflict.

from kibana.exceptions import ConflictError

try:
    # Create space with existing ID
    space = client.spaces.create(
        id="default",  # 'default' space already exists
        name="Default Space"
    )
except ConflictError as e:
    print(f"Conflict: {e.message}")
    print("Resource already exists")

Transport Exceptions¶

TransportError¶

class kibana.TransportError[source]¶

Bases: KibanaException

Transport-level error.

Raised when there’s a problem with the HTTP transport layer, such as connection failures, timeouts, or SSL errors.

Example

>>> try:
...     client = Kibana("http://invalid-host:5601")
...     client.status.get_status()
... except TransportError as e:
...     print(f"Connection error: {e}")

Base class for transport-level errors.

from kibana.exceptions import TransportError

try:
    status = client.status.get_status()
except TransportError as e:
    print(f"Transport error: {e}")

ConnectionError¶

class kibana.ConnectionError[source]¶

Bases: TransportError

Failed to connect to Kibana.

Raised when the client cannot establish a connection to the Kibana server. This may indicate that Kibana is down, the URL is incorrect, or there are network connectivity issues.

Example

>>> try:
...     client = Kibana("http://localhost:9999")  # Wrong port
...     client.status.get_status()
... except ConnectionError as e:
...     print(f"Cannot connect to Kibana: {e}")

Raised when connection to Kibana fails.

from kibana.exceptions import ConnectionError

try:
    client = Kibana("http://nonexistent:5601")
    status = client.status.get_status()
except ConnectionError as e:
    print(f"Cannot connect to Kibana: {e}")
    print("Check that Kibana is running and accessible")

ConnectionTimeout¶

class kibana.ConnectionTimeout[source]¶

Bases: ConnectionError

Connection timed out.

Raised when a connection attempt or request exceeds the configured timeout. This may indicate network issues or an overloaded Kibana server.

Example

>>> client = Kibana("http://localhost:5601", request_timeout=1.0)
>>> try:
...     client.actions.get_all()  # Slow operation
... except ConnectionTimeout as e:
...     print(f"Request timed out: {e}")

Raised when a connection times out.

from kibana.exceptions import ConnectionTimeout

try:
    client = Kibana(
        "http://localhost:5601",
        request_timeout=1  # Very short timeout
    )
    status = client.status.get_status()
except ConnectionTimeout as e:
    print(f"Connection timed out: {e}")

SSLError¶

class kibana.SSLError[source]¶

Bases: ConnectionError

SSL/TLS error.

Raised when there’s a problem with SSL/TLS certificate verification or the secure connection setup.

Example

>>> try:
...     client = Kibana("https://localhost:5601")  # Self-signed cert
...     client.status.get_status()
... except SSLError as e:
...     print(f"SSL error: {e}")

Raised when an SSL/TLS error occurs.

from kibana.exceptions import SSLError

try:
    client = Kibana(
        "https://localhost:5601",
        verify_certs=True
    )
    status = client.status.get_status()
except SSLError as e:
    print(f"SSL error: {e}")
    print("Check SSL certificate configuration")

SerializationError¶

class kibana.SerializationError[source]¶

Bases: KibanaException

Failed to serialize/deserialize data.

Raised when there’s a problem converting data to/from JSON format. This may indicate incompatible data types or corrupted data.

Example

>>> try:
...     # Attempting to serialize non-serializable object
...     client.actions.create(
...         name="Test",
...         connector_type_id=".webhook",
...         config={"callback": lambda x: x}  # Functions can't be serialized
...     )
... except SerializationError as e:
...     print(f"Serialization error: {e}")

Raised when serialization or deserialization fails.

from kibana.exceptions import SerializationError

try:
    # This would raise SerializationError if response is invalid JSON
    result = client.actions.get_all()
except SerializationError as e:
    print(f"Serialization error: {e}")

Space Exceptions¶

SpaceError¶

class kibana.SpaceError[source]¶

Bases: KibanaException

Base exception for space-related errors.

Parent class for all space-specific errors. Use this to catch any space-related exception.

Example

>>> try:
...     client.space("invalid space id!")
... except SpaceError as e:
...     print(f"Space error: {e}")

Base class for space-related errors.

from kibana.exceptions import SpaceError

try:
    connector = client.actions.create(
        name="Test",
        connector_type_id=".webhook",
        config={"url": "https://example.com"},
        space_id="invalid-space"
    )
except SpaceError as e:
    print(f"Space error: {e}")

SpaceNotFoundError¶

class kibana.SpaceNotFoundError(space_id)[source]¶

Bases: SpaceError

Raised when a specified space does not exist.

This exception is raised when attempting to perform operations in a space that doesn’t exist. The space may have been deleted or the ID may be incorrect.

space_id¶: The ID of the space that was not found.

Example

>>> try:
...     client.actions.create(
...         name="Test",
...         connector_type_id=".webhook",
...         config={},
...         space_id="nonexistent"
...     )
... except SpaceNotFoundError as e:
...     print(f"Space '{e.space_id}' not found")
...     print("Please create the space first or check the space ID")

Raised when a specified space does not exist.

Attributes:

space_id (str): The ID of the space that was not found

from kibana.exceptions import SpaceNotFoundError

try:
    connector = client.actions.create(
        name="Test",
        connector_type_id=".webhook",
        config={"url": "https://example.com"},
        space_id="nonexistent"
    )
except SpaceNotFoundError as e:
    print(f"Space not found: {e.space_id}")
    print("Create the space first or use an existing space")

__init__(space_id)[source]¶

Initialize SpaceNotFoundError.

Parameters:: space_id (str) – The ID of the space that was not found.

InvalidSpaceIdError¶

class kibana.InvalidSpaceIdError(space_id)[source]¶

Bases: SpaceError

Raised when a space ID format is invalid.

Space IDs must be URL-friendly (lowercase, alphanumeric, hyphens, underscores). This exception is raised when a space ID contains invalid characters or format.

space_id¶: The invalid space ID that was provided.

Example

>>> try:
...     client.spaces.create(
...         id="Invalid Space!",  # Contains spaces and special chars
...         name="Test Space"
...     )
... except InvalidSpaceIdError as e:
...     print(f"Invalid space ID: {e.space_id}")
...     print("Space IDs must be lowercase and URL-friendly")

Raised when a space ID format is invalid.

Attributes:

space_id (str): The invalid space ID

from kibana.exceptions import InvalidSpaceIdError

try:
    space = client.spaces.create(
        id="Invalid Space!",  # Spaces cannot contain spaces or special chars
        name="Invalid Space"
    )
except InvalidSpaceIdError as e:
    print(f"Invalid space ID: {e.space_id}")
    print("Space IDs must be lowercase alphanumeric with hyphens/underscores")

__init__(space_id)[source]¶

Initialize InvalidSpaceIdError.

Parameters:: space_id (str) – The invalid space ID that was provided.

Error Handling Patterns¶

Specific Exception Handling¶

Handle specific exceptions for fine-grained error handling:

from kibana import Kibana
from kibana.exceptions import (
    NotFoundError,
    ConflictError,
    BadRequestError,
    AuthenticationException,
    SpaceNotFoundError
)

client = Kibana("http://localhost:5601")

try:
    connector = client.actions.create(
        name="My Connector",
        connector_type_id=".webhook",
        config={"url": "https://example.com"},
        space_id="marketing"
    )
except SpaceNotFoundError as e:
    print(f"Space '{e.space_id}' doesn't exist. Creating it...")
    client.spaces.create(id=e.space_id, name="Marketing")
    # Retry operation
except BadRequestError as e:
    print(f"Invalid configuration: {e.message}")
    print(f"Details: {e.body}")
except ConflictError as e:
    print(f"Connector already exists: {e.message}")
except AuthenticationException as e:
    print(f"Authentication failed: {e.message}")
    # Re-authenticate or exit
except NotFoundError as e:
    print(f"Resource not found: {e.message}")

Broad Exception Handling¶

Use base exceptions for broader error handling:

from kibana.exceptions import ApiError, TransportError, KibanaException

try:
    # Perform operations
    result = client.actions.get_all()
except ApiError as e:
    # Handle all API-level errors
    print(f"API error ({e.status_code}): {e.message}")
except TransportError as e:
    # Handle all transport-level errors
    print(f"Transport error: {e}")
except KibanaException as e:
    # Handle any other Kibana errors
    print(f"Kibana error: {e}")

Context Manager Error Handling¶

Proper error handling with context managers:

from kibana import Kibana
from kibana.exceptions import KibanaException

try:
    with Kibana("http://localhost:5601") as client:
        # Perform operations
        status = client.status.get_status()
        print(f"Status: {status.body['status']['overall']['level']}")
except KibanaException as e:
    print(f"Error: {e}")
# Client is automatically closed even if an exception occurs

Async Error Handling¶

Error handling with async client:

from kibana import AsyncKibana
from kibana.exceptions import KibanaException
import asyncio

async def main():
    try:
        async with AsyncKibana("http://localhost:5601") as client:
            # Perform async operations
            status = await client.status.get_status()
            print(f"Status: {status.body['status']['overall']['level']}")
    except KibanaException as e:
        print(f"Error: {e}")

asyncio.run(main())