mypy-boto3-cloudtrail-data Type Annotations

1.42.3 · active · verified Sat Apr 11

Provides type annotations for `boto3`'s CloudTrailDataService, generated by `mypy-boto3-builder`. It enhances static type checking and IDE auto-completion for `boto3` users, currently supporting `boto3` version 1.42.3. Releases follow `boto3`'s release cycle to ensure compatibility.

Warnings

Install

Imports

Quickstart

This example demonstrates how to use the `mypy-boto3-cloudtrail-data` type annotations with a `boto3` client to put audit events into AWS CloudTrail Lake. It showcases explicit type hints for the client and the request/response payloads, which enable static type checking and improved IDE support.

import boto3
from typing import TYPE_CHECKING, List, Dict, Any

if TYPE_CHECKING:
    from mypy_boto3_cloudtrail_data.client import CloudTrailDataServiceClient
    from mypy_boto3_cloudtrail_data.type_defs import (AuditRecordEntryTypeDef, PutAuditEventsRequestRequestTypeDef, PutAuditEventsResponseTypeDef)


def put_cloudtrail_audit_events(event_data_store_id: str, audit_events: List[AuditRecordEntryTypeDef]) -> PutAuditEventsResponseTypeDef:
    """Puts audit events into CloudTrail Lake with type hints."""
    # Boto3 client without type hint - mypy will complain if mypy-boto3 is installed
    # Use TYPE_CHECKING guard for production code to avoid runtime dependency on mypy-boto3
    if TYPE_CHECKING:
        client: CloudTrailDataServiceClient = boto3.client("cloudtrail-data")
    else:
        client = boto3.client("cloudtrail-data")

    # Example request payload with type hint
    request_payload: PutAuditEventsRequestRequestTypeDef = {
        "audit_events": audit_events,
        "EventDataStore": event_data_store_id
    }

    print(f"Putting {len(audit_events)} audit events into {event_data_store_id}...")
    response: PutAuditEventsResponseTypeDef = client.put_audit_events(**request_payload)
    print(f"Result: {response['FailedEntries']}")
    return response


if __name__ == "__main__":
    # Replace with your actual Event Data Store ID
    # For a runnable example, ensure AWS credentials are configured (e.g., via AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
    # and an existing CloudTrail Lake Event Data Store ID is provided.
    sample_event_data_store_id = "your-event-data-store-id" # os.environ.get('CLOUD_TRAIL_DATA_STORE_ID', 'your-event-data-store-id')

    sample_audit_events: List[AuditRecordEntryTypeDef] = [
        {
            "EventID": "example-event-id-1",
            "EventData": "{\"eventVersion\":\"1.08\",\"userIdentity\":{\"type\":\"IAMUser\",\"principalId\":\"AIDAJ45Q7YDR2T3W4XXXB\",\"arn\":\"arn:aws:iam::123456789012:user/Alice\",\"accountId\":\"123456789012\",\"userName\":\"Alice\"},\"eventTime\":\"2026-04-11T12:00:00Z\",\"eventSource\":\"my.application.com\",\"eventName\":\"Login\",\"awsRegion\":\"us-east-1\",\"sourceIPAddress\":\"203.0.113.1\",\"userAgent\":\"Mozilla/5.0\",\"requestParameters\":null,\"responseElements\":null}",
            "EventDataChecksum": "string", # Optional, required if provided for integrity check
            "ExternalID": "string" # Optional
        },
        {
            "EventID": "example-event-id-2",
            "EventData": "{\"eventVersion\":\"1.08\",\"userIdentity\":{\"type\":\"IAMUser\",\"principalId\":\"AIDAJ45Q7YDR2T3W4XXXC\",\"arn\":\"arn:aws:iam::123456789012:user/Bob\",\"accountId\":\"123456789012\",\"userName\":\"Bob\"},\"eventTime\":\"2026-04-11T12:01:00Z\",\"eventSource\":\"my.application.com\",\"eventName\":\"Logout\",\"awsRegion\":\"us-east-1\",\"sourceIPAddress\":\"203.0.113.2\",\"userAgent\":\"Chrome\",\"requestParameters\":null,\"responseElements\":null}"
        }
    ]

    try:
        # Ensure to replace 'your-event-data-store-id' with a valid ID from your AWS account
        if sample_event_data_store_id == 'your-event-data-store-id':
            print("Please replace 'your-event-data-store-id' with a valid CloudTrail Lake Event Data Store ID.")
        else:
            result = put_cloudtrail_audit_events(sample_event_data_store_id, sample_audit_events)
            if result.get("FailedEntries"):
                print(f"Some events failed: {result['FailedEntries']}")
            else:
                print("All events successfully put.")
    except Exception as e:
        print(f"An error occurred: {e}")

view raw JSON →