mypy-boto3-firehose Type Annotations

1.42.3 · active · verified Sat Apr 11

mypy-boto3-firehose provides comprehensive type annotations for the AWS Kinesis Data Firehose service client in `boto3`. It is part of the `mypy-boto3-builder` project, which generates stubs for all `boto3` and `aioboto3` services. The library closely tracks `boto3` versions, with frequent releases that align with AWS SDK updates, ensuring up-to-date type checking for development environments.

Warnings

breaking As of `mypy-boto3-builder` version 8.12.0 (November 8, 2025), Python 3.8 is no longer supported across all `mypy-boto3` packages. Projects must migrate to Python 3.9 or later to receive updates.
Fix: Upgrade your project's Python interpreter to 3.9 or higher. (e.g., `python -m pip install --upgrade python` or update virtual environment/container base image).
breaking The `mypy-boto3-builder` (version 8.12.0 onwards) migrated to PEP 561 compliant packages, which might change how type checkers locate and process stubs. This typically means stubs are automatically discovered, but old or custom `mypy` configurations might need adjustments.
Fix: Ensure `mypy` is configured to automatically discover installed packages. If issues arise, check `mypy` documentation on `PEP 561` and installed packages. No manual `MYPYPATH` for stub-only packages is needed if installed via `pip`.
breaking In `mypy-boto3-builder` version 8.9.0, there were changes to `TypedDict` naming conventions, including shortening names (e.g., `CreateDistributionRequestRequestTypeDef` to `CreateDistributionRequestTypeDef`) and moving the `Extra` postfix. This can break existing explicit `TypedDict` imports or usages.
Fix: Review and update `TypedDict` import paths and names in your codebase according to the new conventions. Consult the specific service's documentation for the correct generated `TypedDict` names.
gotcha While `mypy-boto3` stubs provide robust type checking, some IDEs (like VSCode) or older `mypy` versions might require explicit type annotations for `boto3.client()` calls to provide full autocomplete and type inference due to limitations in function overload support.
Fix: Explicitly annotate the client variable with the correct `mypy-boto3-firehose` client type, e.g., `client: FirehoseClient = boto3.client("firehose")`. This ensures consistent type inference across environments.

Install

pip install mypy-boto3-firehose Install service-specific stubs
pip install boto3-stubs[firehose] Install via boto3-stubs extras

Imports

FirehoseClient
```
from mypy_boto3_firehose import FirehoseClient
```
Directly importing from `boto3.client` will not provide specific type hints for Firehose; `mypy-boto3-firehose` provides the specialized `FirehoseClient`.
PutRecordInputRequestTypeDef
```
from mypy_boto3_firehose.type_defs import PutRecordInputRequestTypeDef
```
For precise type checking of request bodies, use the generated `_type_defs` for specific service operations instead of generic `Dict[str, Any]`.

Quickstart

Demonstrates how to initialize a type-hinted Firehose client and send a record using a `TypedDict` for the request payload, leveraging the provided type annotations. It includes `TYPE_CHECKING` for conditional dependency.

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)

view raw JSON →