Skip to main content

API Reference

This section provides comprehensive documentation for the H2O Audit Trail Python client API.

note

The Python client only supports reading events. Methods create_event() and batch_create_events() return 405 Method Not Allowed errors. To maintain audit trail integrity, event creation is restricted to internal H2O AI Cloud services only.

Authentication Functions

login()

Initializes Audit Trail clients for H2O AI Cloud with automatic configuration discovery.

def login(
environment: Optional[str] = None,
token_provider: Optional[TokenProvider] = None,
platform_token: Optional[str] = None,
config_path: Optional[str] = None,
verify_ssl: bool = True,
ssl_ca_cert: Optional[str] = None,
) -> Clients

Parameters:

  • environment (str, optional): H2O Cloud environment URL (e.g., https://cloud.h2o.ai)
  • token_provider (TokenProvider, optional): Token provider (takes priority over platform_token)
  • platform_token (str, optional): H2O Platform Token
  • config_path (str, optional): Path to H2O CLI config file (defaults to ~/.h2oai/h2o-cli-config.toml)
  • verify_ssl (bool): Enable SSL certificate verification (default: True)
  • ssl_ca_cert (str, optional): Path to CA certificate bundle

Returns:

  • Clients: Object containing initialized clients

Raises:

  • FileNotFoundError: H2O CLI configuration file not found
  • TomlDecodeError: H2O CLI configuration file cannot be processed
  • LookupError: Service endpoint cannot be discovered
  • ConnectionError: Communication with server failed

login_custom()

Initializes Audit Trail clients for custom deployments using OIDC credentials.

def login_custom(
endpoint: str,
refresh_token: str,
issuer_url: str,
client_id: str,
client_secret: Optional[str] = None,
verify_ssl: bool = True,
ssl_ca_cert: Optional[str] = None,
) -> Clients

Parameters:

  • endpoint (str): Audit Trail service endpoint URL
  • refresh_token (str): OIDC refresh token
  • issuer_url (str): OIDC issuer URL
  • client_id (str): OIDC Client ID
  • client_secret (str, optional): OIDC Client Secret
  • verify_ssl (bool): Enable SSL certificate verification (default: True)
  • ssl_ca_cert (str, optional): Path to CA certificate bundle

Returns:

  • Clients: Object containing initialized clients

Clients Class

Clients

Container class that holds initialized client instances.

Attributes:

  • event_client (EventClient): Client for event operations

EventClient Class

EventClient

Main client for managing audit trail events.

class EventClient:
def __init__(
self,
connection_config: ConnectionConfig,
verify_ssl: bool = True,
ssl_ca_cert: Optional[str] = None,
)

create_event()

note

This method returns 405 Method Not Allowed. Event creation is restricted to internal services only.

Creates a single audit event.

def create_event(
self,
event_id: str,
event: Event,
) -> Event

Parameters:

  • event_id (str): UUID4 identifier for the event (e.g., "b096430b-c7a8-47f2-a129-d9ae58cf454f")
  • event (Event): Event object to create

Returns:

  • Event: The created event with server-assigned fields populated

Raises:

  • CustomApiException: Always raises with 405 Method Not Allowed when called from Python client

batch_create_events()

note

This method returns 405 Method Not Allowed. Event creation is restricted to internal services only.

Creates multiple events in a single request with support for partial failures.

def batch_create_events(
self,
requests: List[CreateEventRequest],
) -> BatchCreateEventsResponse

Parameters:

  • requests (List[CreateEventRequest]): List of event creation requests (maximum 1000)

Returns:

  • BatchCreateEventsResponse: Response containing successfully created events and failed requests

Raises:

  • CustomApiException: Always raises with 405 Method Not Allowed when called from Python client

search_events()

Searches for events using filters and pagination.

def search_events(
self,
filter_: Optional[SearchEventsRequestFilter] = None,
page_size: int = 0,
page_token: str = "",
) -> SearchEventsResponse

Parameters:

  • filter_ (SearchEventsRequestFilter, optional): Event filtering criteria
  • page_size (int): Maximum events per page (default: 0, which returns 50; maximum: 1000)
  • page_token (str): Token for pagination (use next_page_token from previous response)

Returns:

  • SearchEventsResponse: Response containing matching events and pagination information

Raises:

  • CustomApiException: API request failed

Data Classes

Event

Represents an audit trail event.

@dataclass
class Event:
event_time: datetime
event_source: str
action: str
read_only: bool
status: Status
principal: str
source_ip_address: str
name: str = ""
login_principal: str = ""
receive_time: Optional[datetime] = None
resource: str = ""
request_parameters: Dict[str, str] = field(default_factory=dict)
user_agent: str = ""
metadata: Dict[str, str] = field(default_factory=dict)
workspace: str = ""

Fields:

  • event_time (datetime): When the event occurred
  • event_source (str): Container image name (e.g., "h2oai-enginemanager-server")
  • action (str): Requested action (e.g., "actions/enginemanager/daiEngines/CREATE")
  • read_only (bool): Whether the action is read-only
  • status (Status): Request status with code and optional error details
  • principal (str): Authenticated principal identifier (e.g., "users/user-id")
  • source_ip_address (str): Source IP address (private IPs are redacted to "private")
  • name (str): Resource name in format events/{event}
  • login_principal (str): User-friendly identifier for authentication
  • receive_time (datetime, optional): When the event was received
  • resource (str): Target resource (e.g., "//engine-manager/workspaces/id/daiEngines/name")
  • request_parameters (Dict[str, str]): Request parameters (may be filtered)
  • user_agent (str): Request user agent
  • metadata (Dict[str, str]): Service-specific data
  • workspace (str): Related workspace in format workspaces/*

Status

Represents the status of a request.

@dataclass
class Status:
code: int
message: str = ""

Fields:

  • code (int): Status code (0=OK, 1=CANCELLED, 2=UNKNOWN, 3=INVALID_ARGUMENT, etc.)
  • message (str): Optional error message

SearchEventsRequestFilter

Filter criteria for event searches.

@dataclass
class SearchEventsRequestFilter:
start_event_time: Optional[datetime] = None
end_event_time: Optional[datetime] = None
event_source_exact: Optional[str] = None
event_source_regex: Optional[str] = None
resource_exact: Optional[str] = None
resource_regex: Optional[str] = None
read_only: Optional[bool] = None
status_code_regex: Optional[str] = None
principal_exact: Optional[str] = None
principal_regex: Optional[str] = None
source_ip_address_exact: Optional[str] = None
source_ip_address_regex: Optional[str] = None
metadata_exact: Dict[str, str] = field(default_factory=dict)
metadata_regex: Dict[str, str] = field(default_factory=dict)
workspace_exact: Optional[str] = None
workspace_regex: Optional[str] = None
action_exact: Optional[str] = None
action_regex: Optional[str] = None

Filter Types:

  • Time filters: start_event_time, end_event_time
  • Exact match filters: Fields ending with _exact
  • Regex filters: Fields ending with _regex (must follow Google RE2 syntax)
  • Metadata filters: metadata_exact (key-value pairs), metadata_regex (key-regex pairs)

SearchEventsResponse

Response from event search operations.

@dataclass
class SearchEventsResponse:
events: List[Event]
next_page_token: str
searched_until_time: datetime

Fields:

  • events (List[Event]): Matching events
  • next_page_token (str): Token for next page (empty if no more results)
  • searched_until_time (datetime): Search progress indicator (for informational purposes only)

CreateEventRequest

Request object for batch event creation.

@dataclass
class CreateEventRequest:
event_id: str
event: Event

Fields:

  • event_id (str): UUID4 identifier for the event
  • event (Event): Event object to create

BatchCreateEventsResponse

Response from batch event creation.

@dataclass
class BatchCreateEventsResponse:
events: List[Event]
failed_requests: Dict[int, RPCStatus]

Fields:

  • events (List[Event]): Successfully created events
  • failed_requests (Dict[int, RPCStatus]): Failed requests mapping index to error details

Exceptions

CustomApiException

Custom exception wrapper for API errors.

class CustomApiException(Exception):
def __init__(self, api_exception: ApiException)

Wraps underlying ApiException from the generated client with additional context.


Feedback