Type Annotations for aiobotocore Kinesis

3.4.0 · active · verified Thu Apr 16

This library provides PEP 561 compatible type annotations for the `aiobotocore` Kinesis service. It enhances development with static type checking, auto-completion, and improved code readability when working with the asynchronous AWS client. The `types-aiobotocore` project, generated by `mypy-boto3-builder`, has a rapid release cadence, ensuring type definitions are kept up-to-date with `aiobotocore` versions.

Common errors

```
ModuleNotFoundError: No module named 'aiobotocore'
```
cause The `types-aiobotocore-kinesis` package only provides type hints. The actual `aiobotocore` library, which contains the runtime code, is not installed.

fix Install the `aiobotocore` library: `pip install aiobotocore` (or `pip install types-aiobotocore-kinesis aiobotocore` to get both).
```
error: Missing type stub for 'aiobotocore'
```
cause You have installed `aiobotocore` but not its corresponding type stubs, leading type checkers (like mypy) to report missing type information.

fix Install the Kinesis-specific type stubs: `pip install types-aiobotocore-kinesis`. If you need stubs for other services, consider `types-aiobotocore[essential]` or specific service packages.
```
error: Argument "StreamName" to "put_record" of "KinesisClient" has incompatible type "int"; expected "str"
```
cause A Kinesis client method was called with an argument of the wrong type, detected by the type stubs.

fix Consult the `types-aiobotocore-kinesis` documentation or your IDE's auto-completion for the correct argument types (e.g., `StreamName` expects a string).

Warnings

breaking Support for Python 3.8 has been removed in `mypy-boto3-builder` version 8.12.0, which generates these type stubs. Packages generated with this builder version and later will only support Python 3.9 and higher.
Fix: Upgrade your Python environment to 3.9 or newer. If stuck on Python 3.8, use an older version of `types-aiobotocore-kinesis` (e.g., <3.4.0) that was generated with `mypy-boto3-builder` <8.12.0.
breaking The `mypy-boto3-builder` migrated to PEP 561 package format in version 8.12.0. While largely internal, this change might affect how some tools discover or interact with the stub packages.
Fix: Ensure your type checker and IDE are updated to recent versions that fully support PEP 561 packages. No direct code change is typically required for end-users.
gotcha This package provides *only* type annotations. The actual `aiobotocore` library must be installed separately for your code to run at runtime. Failure to install `aiobotocore` will result in a `ModuleNotFoundError`.
Fix: Always install both the type stubs and the runtime library: `pip install types-aiobotocore-kinesis aiobotocore`.
gotcha Users of PyCharm may experience slow performance or high CPU usage due to how PyCharm's type checker handles `Literal` overloads in these generated stubs. It is recommended to use `mypy` or `pyright` for type checking, or to consider `types-aiobotocore-lite` packages as a workaround if PyCharm's internal type checker is preferred.
Fix: Disable PyCharm's internal type checker and use an external tool like `mypy` or `pyright`. Alternatively, install `types-aiobotocore-lite` for the specific service (if available) which offers a lighter set of annotations.

Install

pip install types-aiobotocore-kinesis aiobotocore Install type stubs and runtime library

Imports

KinesisClient
wrong:
```
from aiobotocore.kinesis.client import KinesisClient
```
correct:
```
from types_aiobotocore_kinesis.client import KinesisClient
```
Type stubs are imported from their dedicated `types-` package, not directly from the runtime library.

PutRecordRequestRequestTypeDef

wrong:

from aiobotocore.kinesis.type_defs import PutRecordRequestRequestTypeDef

correct:

from types_aiobotocore_kinesis.type_defs import PutRecordRequestRequestTypeDef

TypedDicts and other type definitions are also imported from the `types-` package.

Quickstart

This quickstart demonstrates how to use `types-aiobotocore-kinesis` with `aiobotocore` to put a record into an Amazon Kinesis stream. It shows how to explicitly type the Kinesis client and the request parameters using `TypedDicts` for improved type checking and IDE auto-completion. This example requires `aiobotocore` and valid AWS credentials configured for the specified region.

import asyncio
from aiobotocore.session import get_session
from types_aiobotocore_kinesis.client import KinesisClient
from types_aiobotocore_kinesis.type_defs import PutRecordRequestRequestTypeDef
import os

async def put_kinesis_record(stream_name: str, data: str, partition_key: str):
    session = get_session()
    async with session.create_client("kinesis", region_name=os.environ.get('AWS_REGION', 'us-east-1')) as client:  # type: KinesisClient
        request_params: PutRecordRequestRequestTypeDef = {
            "StreamName": stream_name,
            "Data": data.encode('utf-8'),
            "PartitionKey": partition_key,
        }
        response = await client.put_record(**request_params)
        print(f"Shard ID: {response['ShardId']}, Sequence Number: {response['SequenceNumber']}")

async def main():
    # Example usage: Replace with your stream name, data, and partition key
    # Ensure AWS_REGION environment variable is set or passed to create_client
    # Also ensure AWS credentials are configured.
    await put_kinesis_record("my-test-stream", "Hello Kinesis!", "my-partition-key")

if __name__ == "__main__":
    # For demonstration, setting dummy env var if not present
    if 'AWS_REGION' not in os.environ:
        os.environ['AWS_REGION'] = 'us-east-1'
    asyncio.run(main())

view raw JSON →