mypy-boto3-firehose

import boto3 import os from typing import TYPE_CHECKING if TYPE_CHECKING: from mypy_boto3_firehose import FirehoseClient from mypy_boto3_firehose.type_defs import PutRecordInputRequestTypeDef # Configure AWS credentials (e.g., via environment variables or AWS CLI config) # For example, ensure AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION_NAME are set. def send_to_firehose(delivery_stream_name: str, data: str) -> None: """Sends data to an AWS Kinesis Firehose delivery stream with type hints.""" # Explicit type annotation for the client (recommended for some IDEs/type checkers) client: FirehoseClient = boto3.client( "firehose", region_name=os.environ.get('AWS_REGION_NAME', 'us-east-1'), aws_access_key_id=os.environ.get('AWS_ACCESS_KEY_ID', ''), aws_secret_access_key=os.environ.get('AWS_SECRET_ACCESS_KEY', ''), ) # Use a TypedDict for the request payload for full type checking record_data: PutRecordInputRequestTypeDef = { "DeliveryStreamName": delivery_stream_name, "Record": {"Data": data.encode('utf-8')}, } try: response = client.put_record(**record_data) print(f"Successfully sent record. RecordId: {response['RecordId']}") except Exception as e: print(f"Error sending record to Firehose: {e}") if __name__ == "__main__": # Example usage (replace with your actual stream name and data) firehose_stream = "YourFirehoseDeliveryStreamName" message = "Hello, Firehose! This is a typed record." send_to_firehose(firehose_stream, message)