Type Annotations for Boto3 Kinesis

Warnings

• breaking Support for Python 3.8 was removed in `mypy-boto3-builder` version 8.12.0 (and consequently affects `mypy-boto3-kinesis`). Users on Python 3.8 should upgrade their Python version.
  Fix: Upgrade Python to 3.9 or higher.
• breaking In `mypy-boto3-builder` version 8.9.0, some `TypeDef` names were shortened (e.g., `CreateDistributionRequestRequestTypeDef` became `CreateDistributionRequestTypeDef`). While this example is for CloudFront, similar changes might affect Kinesis-specific `TypeDefs` if they followed the same pattern.
  Fix: Review your code for any custom `TypeDef` imports from `mypy_boto3_kinesis.type_defs` and update names if they reflect the shortened pattern.
• gotcha This package provides *type annotations* for Boto3; it does not replace the actual `boto3` library, which must be installed separately and is required at runtime for your application to function.
  Fix: Always ensure `boto3` is installed alongside `mypy-boto3-kinesis` or `boto3-stubs[kinesis]`.
• gotcha When using `boto3.client()` or `boto3.session.client()` without installing `boto3-stubs` (i.e., only `mypy-boto3-kinesis`), explicit type annotations are often necessary for optimal IDE auto-completion and static analysis. For example: `client: KinesisClient = boto3.client("kinesis")`.
  Fix: For best type inference without explicit annotations, install the full `boto3-stubs` package with the Kinesis extra: `pip install 'boto3-stubs[kinesis]'`.
• gotcha For Pylint compatibility and to avoid `undefined-variable` warnings when using `TYPE_CHECKING` for conditional imports, you might need to assign `object` to your client types in the non-`TYPE_CHECKING` block.
  Fix: Use `if TYPE_CHECKING: from mypy_boto3_kinesis.client import KinesisClient else: KinesisClient = object` in your code.

Install

• pip install boto3 mypy-boto3-kinesis Standalone Kinesis type annotations
• pip install boto3 'boto3-stubs[kinesis]' Recommended: Boto3-stubs with Kinesis extra

Imports

• KinesisClient

  from mypy_boto3_kinesis.client import KinesisClient

  Import for the Kinesis client type annotation.
• StreamExistsWaiter

  from mypy_boto3_kinesis.waiters import StreamExistsWaiter

  Import for Kinesis StreamExistsWaiter type annotation.
• ConsumerStatusType

  from mypy_boto3_kinesis.literals import ConsumerStatusType

  Import for Kinesis literal types, useful for strict type checking of predefined string values.
• ListStreamsOutputTypeDef

  from mypy_boto3_kinesis.type_defs import ListStreamsOutputTypeDef

  Import for Kinesis TypedDict definitions, for type-hinting API response structures.

Quickstart

This quickstart demonstrates how to initialize a Boto3 Kinesis client with type annotations and list available Kinesis streams. Explicit type hints for the client and response structures provide static analysis benefits. Ensure AWS credentials and region are configured in your environment for the code to run successfully.

import boto3
from mypy_boto3_kinesis.client import KinesisClient
from mypy_boto3_kinesis.type_defs import ListStreamsOutputTypeDef
import os

def list_kinesis_streams() -> ListStreamsOutputTypeDef:
    # Ensure boto3 is configured, e.g., via AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY,
    # and AWS_REGION environment variables or ~/.aws/credentials
    # The 'client' variable is explicitly type-hinted for mypy/IDE benefit
    client: KinesisClient = boto3.client("kinesis", region_name=os.environ.get('AWS_REGION', 'us-east-1'))
    
    response: ListStreamsOutputTypeDef = client.list_streams()
    print("Kinesis Streams:")
    for stream_name in response.get("StreamNames", []):
        print(f"- {stream_name}")
    return response

if __name__ == "__main__":
    # Example usage requires AWS credentials and region to be set up
    # For demonstration, ensure environment variables are set or remove this check
    if not os.environ.get('AWS_ACCESS_KEY_ID') or not os.environ.get('AWS_SECRET_ACCESS_KEY'):
        print("Please set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables.")
        print("Also, optionally set AWS_REGION (defaults to 'us-east-1').")
    else:
        try:
            list_kinesis_streams()
        except Exception as e:
            print(f"An error occurred: {e}")