rboto-cloudtrail

rboto is an independent project maintained by Ferrumio. It is not affiliated with, sponsored by, or endorsed by Amazon Web Services (AWS).

Preview (Alpha). rboto is under active development. APIs may change between releases.

rboto provides an async-only, typed Python client for AWS CloudTrail, powered by Rust. This package contains CloudTrailClient and 60 operations.

Installation

Requires Python 3.12 or newer and a native wheel for your Python version and platform.

python -m pip install "rboto[cloudtrail]"

You can also install rboto-cloudtrail directly; it installs the matching rboto package. Only the selected service is installed.

These are the package names to use when a release is available on your package index. If your index has no release yet, follow the source installation instructions.

Usage

Configure credentials through your environment, shared profile, or workload role. The example uses us-east-1; select a region where your service and resources are available. Resource names, identifiers, input files, and model IDs in the example are placeholders to replace with your own.

import asyncio

from rboto import CloudTrailClient


async def main() -> None:
    client = CloudTrailClient(region="us-east-1")
    response = await client.list_trails()
    for trail in response.trails or []:
        print(trail.name, trail.trail_arn)


if __name__ == "__main__":
    asyncio.run(main())

In an application with an event loop, call await main() instead of asyncio.run(main()). Reuse the client across calls. Constructing a client does not send a request.

API conventions

• Import the client with from rboto import CloudTrailClient.

• Every operation is asynchronous: use await client.method(...).

• Operation arguments are keyword-only and use the Python names listed below.

• Structured inputs use typed dictionaries and tagged unions from rboto_cloudtrail.types.

• Regular responses expose typed attributes; response.to_dict() converts them to a dictionary.

• Event-stream responses may be typed dictionaries containing async iterators. Follow the declared return type; consume event streams with async for.

• Byte streams are consumed asynchronously. await stream.read() buffers the complete body; iterate the stream when processing a large response.

• Direct operation calls return one page.

• Use async for page in client.paginators.method(...) for supported paginated operations. Paginators accept the same typed arguments as the corresponding client method.

• Catch modeled exceptions from rboto_cloudtrail.exceptions. Their error_code, message, and operation_name identify service failures.

Clients also accept profile, endpoint_url, or shared Session settings. See configuration, errors, and streaming.

Operations

This is the complete operation catalog for this generated package. “Required arguments” lists arguments without defaults in the Python signature. Service-side conditions may require additional arguments. Each method links to its complete signature, including optional arguments. Nested input definitions are in rboto_cloudtrail.types; native response definitions are in the package's _native.pyi.

Method	Required arguments	Return type
add_tags	resource_id: str, tags_list: Sequence[Tag]	AddTagsOutput
cancel_query	query_id: str	CancelQueryOutput
create_channel	name: str, source: str, destinations: Sequence[Destination]	CreateChannelOutput
create_dashboard	name: str	CreateDashboardOutput
create_event_data_store	name: str	CreateEventDataStoreOutput
create_trail	name: str, s3_bucket_name: str	CreateTrailOutput
delete_channel	channel: str	DeleteChannelOutput
delete_dashboard	dashboard_id: str	DeleteDashboardOutput
delete_event_data_store	event_data_store: str	DeleteEventDataStoreOutput
delete_resource_policy	resource_arn: str	DeleteResourcePolicyOutput
delete_trail	name: str	DeleteTrailOutput
deregister_organization_delegated_admin	delegated_admin_account_id: str	DeregisterOrganizationDelegatedAdminOutput
describe_query	None	DescribeQueryOutput
describe_trails	None	DescribeTrailsOutput
disable_federation	event_data_store: str	DisableFederationOutput
enable_federation	event_data_store: str, federation_role_arn: str	EnableFederationOutput
generate_query	event_data_stores: Sequence[str], prompt: str	GenerateQueryOutput
get_channel	channel: str	GetChannelOutput
get_dashboard	dashboard_id: str	GetDashboardOutput
get_event_configuration	None	GetEventConfigurationOutput
get_event_data_store	event_data_store: str	GetEventDataStoreOutput
get_event_selectors	trail_name: str	GetEventSelectorsOutput
get_import	import_id: str	GetImportOutput
get_insight_selectors	None	GetInsightSelectorsOutput
get_query_results	query_id: str	GetQueryResultsOutput
get_resource_policy	resource_arn: str	GetResourcePolicyOutput
get_trail	name: str	GetTrailOutput
get_trail_status	name: str	GetTrailStatusOutput
list_channels	None	ListChannelsOutput
list_dashboards	None	ListDashboardsOutput
list_event_data_stores	None	ListEventDataStoresOutput
list_import_failures	import_id: str	ListImportFailuresOutput
list_imports	None	ListImportsOutput
list_insights_data	insight_source: str, data_type: ListInsightsDataType	ListInsightsDataOutput
list_insights_metric_data	event_source: str, event_name: str, insight_type: InsightType	ListInsightsMetricDataOutput
list_public_keys	None	ListPublicKeysOutput
list_queries	event_data_store: str	ListQueriesOutput
list_tags	resource_id_list: Sequence[str]	ListTagsOutput
list_trails	None	ListTrailsOutput
lookup_events	None	LookupEventsOutput
put_event_configuration	None	PutEventConfigurationOutput
put_event_selectors	trail_name: str	PutEventSelectorsOutput
put_insight_selectors	insight_selectors: Sequence[InsightSelector]	PutInsightSelectorsOutput
put_resource_policy	resource_arn: str, resource_policy: str	PutResourcePolicyOutput
register_organization_delegated_admin	member_account_id: str	RegisterOrganizationDelegatedAdminOutput
remove_tags	resource_id: str, tags_list: Sequence[Tag]	RemoveTagsOutput
restore_event_data_store	event_data_store: str	RestoreEventDataStoreOutput
search_sample_queries	search_phrase: str	SearchSampleQueriesOutput
start_dashboard_refresh	dashboard_id: str	StartDashboardRefreshOutput
start_event_data_store_ingestion	event_data_store: str	StartEventDataStoreIngestionOutput
start_import	None	StartImportOutput
start_logging	name: str	StartLoggingOutput
start_query	None	StartQueryOutput
stop_event_data_store_ingestion	event_data_store: str	StopEventDataStoreIngestionOutput
stop_import	import_id: str	StopImportOutput
stop_logging	name: str	StopLoggingOutput
update_channel	channel: str	UpdateChannelOutput
update_dashboard	dashboard_id: str	UpdateDashboardOutput
update_event_data_store	event_data_store: str	UpdateEventDataStoreOutput
update_trail	name: str	UpdateTrailOutput

Complete signatures

add_tags

async def add_tags(
    *,
    resource_id: str,
    tags_list: Sequence[Tag],
) -> AddTagsOutput: ...

cancel_query

async def cancel_query(
    *,
    query_id: str,
    event_data_store: str | None = None,
    event_data_store_owner_account_id: str | None = None,
) -> CancelQueryOutput: ...

create_channel

async def create_channel(
    *,
    name: str,
    source: str,
    destinations: Sequence[Destination],
    tags: Sequence[Tag] | None = None,
) -> CreateChannelOutput: ...

create_dashboard

async def create_dashboard(
    *,
    name: str,
    refresh_schedule: RefreshSchedule | None = None,
    tags_list: Sequence[Tag] | None = None,
    termination_protection_enabled: bool | None = None,
    widgets: Sequence[RequestWidget] | None = None,
) -> CreateDashboardOutput: ...

create_event_data_store

async def create_event_data_store(
    *,
    name: str,
    advanced_event_selectors: Sequence[AdvancedEventSelector] | None = None,
    multi_region_enabled: bool | None = None,
    organization_enabled: bool | None = None,
    retention_period: int | None = None,
    termination_protection_enabled: bool | None = None,
    tags_list: Sequence[Tag] | None = None,
    kms_key_id: str | None = None,
    start_ingestion: bool | None = None,
    billing_mode: BillingMode | None = None,
) -> CreateEventDataStoreOutput: ...

create_trail

async def create_trail(
    *,
    name: str,
    s3_bucket_name: str,
    s3_key_prefix: str | None = None,
    sns_topic_name: str | None = None,
    include_global_service_events: bool | None = None,
    is_multi_region_trail: bool | None = None,
    enable_log_file_validation: bool | None = None,
    cloud_watch_logs_log_group_arn: str | None = None,
    cloud_watch_logs_role_arn: str | None = None,
    kms_key_id: str | None = None,
    is_organization_trail: bool | None = None,
    tags_list: Sequence[Tag] | None = None,
    recursive_logging: bool | None = None,
) -> CreateTrailOutput: ...

delete_channel

async def delete_channel(
    *,
    channel: str,
) -> DeleteChannelOutput: ...

delete_dashboard

async def delete_dashboard(
    *,
    dashboard_id: str,
) -> DeleteDashboardOutput: ...

delete_event_data_store

async def delete_event_data_store(
    *,
    event_data_store: str,
) -> DeleteEventDataStoreOutput: ...

delete_resource_policy

async def delete_resource_policy(
    *,
    resource_arn: str,
) -> DeleteResourcePolicyOutput: ...

delete_trail

async def delete_trail(
    *,
    name: str,
) -> DeleteTrailOutput: ...

deregister_organization_delegated_admin

async def deregister_organization_delegated_admin(
    *,
    delegated_admin_account_id: str,
) -> DeregisterOrganizationDelegatedAdminOutput: ...

describe_query

async def describe_query(
    *,
    event_data_store: str | None = None,
    query_id: str | None = None,
    query_alias: str | None = None,
    refresh_id: str | None = None,
    event_data_store_owner_account_id: str | None = None,
) -> DescribeQueryOutput: ...

describe_trails

async def describe_trails(
    *,
    trail_name_list: Sequence[str] | None = None,
    include_shadow_trails: bool | None = None,
) -> DescribeTrailsOutput: ...

disable_federation

async def disable_federation(
    *,
    event_data_store: str,
) -> DisableFederationOutput: ...

enable_federation

async def enable_federation(
    *,
    event_data_store: str,
    federation_role_arn: str,
) -> EnableFederationOutput: ...

generate_query

async def generate_query(
    *,
    event_data_stores: Sequence[str],
    prompt: str,
) -> GenerateQueryOutput: ...

get_channel

async def get_channel(
    *,
    channel: str,
) -> GetChannelOutput: ...

get_dashboard

async def get_dashboard(
    *,
    dashboard_id: str,
) -> GetDashboardOutput: ...

get_event_configuration

async def get_event_configuration(
    *,
    trail_name: str | None = None,
    event_data_store: str | None = None,
) -> GetEventConfigurationOutput: ...

get_event_data_store

async def get_event_data_store(
    *,
    event_data_store: str,
) -> GetEventDataStoreOutput: ...

get_event_selectors

async def get_event_selectors(
    *,
    trail_name: str,
) -> GetEventSelectorsOutput: ...

get_import

async def get_import(
    *,
    import_id: str,
) -> GetImportOutput: ...

get_insight_selectors

async def get_insight_selectors(
    *,
    trail_name: str | None = None,
    event_data_store: str | None = None,
) -> GetInsightSelectorsOutput: ...

get_query_results

async def get_query_results(
    *,
    query_id: str,
    event_data_store: str | None = None,
    next_token: str | None = None,
    max_query_results: int | None = None,
    event_data_store_owner_account_id: str | None = None,
) -> GetQueryResultsOutput: ...

Also available as client.paginators.get_query_results(...), returning an async iterator of the response type above.

get_resource_policy

async def get_resource_policy(
    *,
    resource_arn: str,
) -> GetResourcePolicyOutput: ...

get_trail

async def get_trail(
    *,
    name: str,
) -> GetTrailOutput: ...

get_trail_status

async def get_trail_status(
    *,
    name: str,
) -> GetTrailStatusOutput: ...

list_channels

async def list_channels(
    *,
    max_results: int | None = None,
    next_token: str | None = None,
) -> ListChannelsOutput: ...

Also available as client.paginators.list_channels(...), returning an async iterator of the response type above.

list_dashboards

async def list_dashboards(
    *,
    name_prefix: str | None = None,
    type: DashboardType | None = None,
    next_token: str | None = None,
    max_results: int | None = None,
) -> ListDashboardsOutput: ...

list_event_data_stores

async def list_event_data_stores(
    *,
    next_token: str | None = None,
    max_results: int | None = None,
) -> ListEventDataStoresOutput: ...

Also available as client.paginators.list_event_data_stores(...), returning an async iterator of the response type above.

list_import_failures

async def list_import_failures(
    *,
    import_id: str,
    max_results: int | None = None,
    next_token: str | None = None,
) -> ListImportFailuresOutput: ...

Also available as client.paginators.list_import_failures(...), returning an async iterator of the response type above.

list_imports

async def list_imports(
    *,
    max_results: int | None = None,
    destination: str | None = None,
    import_status: ImportStatus | None = None,
    next_token: str | None = None,
) -> ListImportsOutput: ...

Also available as client.paginators.list_imports(...), returning an async iterator of the response type above.

list_insights_data

async def list_insights_data(
    *,
    insight_source: str,
    data_type: ListInsightsDataType,
    dimensions: dict[ListInsightsDataDimensionKey, str] | None = None,
    start_time: str | None = None,
    end_time: str | None = None,
    max_results: int | None = None,
    next_token: str | None = None,
) -> ListInsightsDataOutput: ...

Also available as client.paginators.list_insights_data(...), returning an async iterator of the response type above.

list_insights_metric_data

async def list_insights_metric_data(
    *,
    event_source: str,
    event_name: str,
    insight_type: InsightType,
    trail_name: str | None = None,
    error_code: str | None = None,
    start_time: str | None = None,
    end_time: str | None = None,
    period: int | None = None,
    data_type: InsightsMetricDataType | None = None,
    max_results: int | None = None,
    next_token: str | None = None,
) -> ListInsightsMetricDataOutput: ...

Also available as client.paginators.list_insights_metric_data(...), returning an async iterator of the response type above.

list_public_keys

async def list_public_keys(
    *,
    start_time: str | None = None,
    end_time: str | None = None,
    next_token: str | None = None,
) -> ListPublicKeysOutput: ...

Also available as client.paginators.list_public_keys(...), returning an async iterator of the response type above.

list_queries

async def list_queries(
    *,
    event_data_store: str,
    next_token: str | None = None,
    max_results: int | None = None,
    start_time: str | None = None,
    end_time: str | None = None,
    query_status: QueryStatus | None = None,
) -> ListQueriesOutput: ...

Also available as client.paginators.list_queries(...), returning an async iterator of the response type above.

list_tags

async def list_tags(
    *,
    resource_id_list: Sequence[str],
    next_token: str | None = None,
) -> ListTagsOutput: ...

Also available as client.paginators.list_tags(...), returning an async iterator of the response type above.

list_trails

async def list_trails(
    *,
    next_token: str | None = None,
) -> ListTrailsOutput: ...

Also available as client.paginators.list_trails(...), returning an async iterator of the response type above.

lookup_events

async def lookup_events(
    *,
    lookup_attributes: Sequence[LookupAttribute] | None = None,
    start_time: str | None = None,
    end_time: str | None = None,
    event_category: EventCategory | None = None,
    max_results: int | None = None,
    next_token: str | None = None,
) -> LookupEventsOutput: ...

Also available as client.paginators.lookup_events(...), returning an async iterator of the response type above.

put_event_configuration

async def put_event_configuration(
    *,
    trail_name: str | None = None,
    event_data_store: str | None = None,
    max_event_size: MaxEventSize | None = None,
    context_key_selectors: Sequence[ContextKeySelector] | None = None,
    aggregation_configurations: Sequence[AggregationConfiguration] | None = None,
) -> PutEventConfigurationOutput: ...

put_event_selectors

async def put_event_selectors(
    *,
    trail_name: str,
    event_selectors: Sequence[EventSelector] | None = None,
    advanced_event_selectors: Sequence[AdvancedEventSelector] | None = None,
) -> PutEventSelectorsOutput: ...

put_insight_selectors

async def put_insight_selectors(
    *,
    insight_selectors: Sequence[InsightSelector],
    trail_name: str | None = None,
    event_data_store: str | None = None,
    insights_destination: str | None = None,
) -> PutInsightSelectorsOutput: ...

put_resource_policy

async def put_resource_policy(
    *,
    resource_arn: str,
    resource_policy: str,
) -> PutResourcePolicyOutput: ...

register_organization_delegated_admin

async def register_organization_delegated_admin(
    *,
    member_account_id: str,
) -> RegisterOrganizationDelegatedAdminOutput: ...

remove_tags

async def remove_tags(
    *,
    resource_id: str,
    tags_list: Sequence[Tag],
) -> RemoveTagsOutput: ...

restore_event_data_store

async def restore_event_data_store(
    *,
    event_data_store: str,
) -> RestoreEventDataStoreOutput: ...

search_sample_queries

async def search_sample_queries(
    *,
    search_phrase: str,
    max_results: int | None = None,
    next_token: str | None = None,
) -> SearchSampleQueriesOutput: ...

start_dashboard_refresh

async def start_dashboard_refresh(
    *,
    dashboard_id: str,
    query_parameter_values: dict[str, str] | None = None,
) -> StartDashboardRefreshOutput: ...

start_event_data_store_ingestion

async def start_event_data_store_ingestion(
    *,
    event_data_store: str,
) -> StartEventDataStoreIngestionOutput: ...

start_import

async def start_import(
    *,
    destinations: Sequence[str] | None = None,
    import_source: ImportSource | None = None,
    start_event_time: str | None = None,
    end_event_time: str | None = None,
    import_id: str | None = None,
) -> StartImportOutput: ...

start_logging

async def start_logging(
    *,
    name: str,
) -> StartLoggingOutput: ...

start_query

async def start_query(
    *,
    query_statement: str | None = None,
    delivery_s3_uri: str | None = None,
    query_alias: str | None = None,
    query_parameters: Sequence[str] | None = None,
    event_data_store_owner_account_id: str | None = None,
) -> StartQueryOutput: ...

stop_event_data_store_ingestion

async def stop_event_data_store_ingestion(
    *,
    event_data_store: str,
) -> StopEventDataStoreIngestionOutput: ...

stop_import

async def stop_import(
    *,
    import_id: str,
) -> StopImportOutput: ...

stop_logging

async def stop_logging(
    *,
    name: str,
) -> StopLoggingOutput: ...

update_channel

async def update_channel(
    *,
    channel: str,
    destinations: Sequence[Destination] | None = None,
    name: str | None = None,
) -> UpdateChannelOutput: ...

update_dashboard

async def update_dashboard(
    *,
    dashboard_id: str,
    widgets: Sequence[RequestWidget] | None = None,
    refresh_schedule: RefreshSchedule | None = None,
    termination_protection_enabled: bool | None = None,
) -> UpdateDashboardOutput: ...

update_event_data_store

async def update_event_data_store(
    *,
    event_data_store: str,
    name: str | None = None,
    advanced_event_selectors: Sequence[AdvancedEventSelector] | None = None,
    multi_region_enabled: bool | None = None,
    organization_enabled: bool | None = None,
    retention_period: int | None = None,
    termination_protection_enabled: bool | None = None,
    kms_key_id: str | None = None,
    billing_mode: BillingMode | None = None,
) -> UpdateEventDataStoreOutput: ...

update_trail

async def update_trail(
    *,
    name: str,
    s3_bucket_name: str | None = None,
    s3_key_prefix: str | None = None,
    sns_topic_name: str | None = None,
    include_global_service_events: bool | None = None,
    is_multi_region_trail: bool | None = None,
    enable_log_file_validation: bool | None = None,
    cloud_watch_logs_log_group_arn: str | None = None,
    cloud_watch_logs_role_arn: str | None = None,
    kms_key_id: str | None = None,
    is_organization_trail: bool | None = None,
    recursive_logging: bool | None = None,
) -> UpdateTrailOutput: ...

Links

• Service guide
• Repository
• Issues
• Releases

Licensed under Apache-2.0.