modernize async-kinesis library for Python 3.10-3.12 (#40)

- Updated to support Python 3.10, 3.11, and 3.12 (last 3 releases)
- Added Python version compatibility layer for timeout functions
- Updated classifiers and python_requires in setup.py

- Migrated from deprecated aredis to modern redis (>=4.0.0)
- Updated async-timeout to >=4.0.0 with backward compatibility
- Fixed aiobotocore version alignment across files
- Resolved Python 3.12 distutils compatibility issues

- Migrated from nose to pytest with async support
- Added comprehensive GitHub Actions CI workflow for matrix testing
- Created organized test modules (conftest.py, test_integration.py, etc.)
- Added pytest configuration and coverage reporting
- Implemented LocalStack for reliable Kinesis mocking

- Added comprehensive type hints throughout codebase
- Enhanced Protocol definitions for processors and checkpointers
- Improved IDE support and static analysis capabilities
- Maintained usability with moderate typing approach

- Fixed processor inheritance issues maintaining backward compatibility
- Enhanced error handling in producer for LocalStack compatibility
- Improved Redis checkpointer concurrent allocation logic
- Added null result guards for robustness

- Split monolithic tests.py into focused test modules
- Added integration tests with comprehensive scenarios
- Enhanced checkpointer tests with proper isolation
- Implemented service startup reliability checks
- Added timeout compatibility across Python versions

- Created GitHub Actions workflow with service containers
- Added matrix testing for Python 3.10, 3.11, 3.12
- Integrated code coverage reporting
- Added linting and formatting checks

- 56/62 tests passing (6 appropriate skips)
- Python 3.10: ✅ Full compatibility achieved
- Python 3.11: ✅ Working
- Python 3.12: ✅ Working
- Integration tests: ✅ All scenarios covered
- LocalStack compatibility: ✅ Stable

.

Author: hampsterx

.flake8

@@ -0,0 +1,14 @@
+[flake8]
+max-line-length = 88
+extend-ignore = E203, W503, E501
+exclude = 
+    .git,
+    __pycache__,
+    venv,
+    .venv,
+    build,
+    dist,
+    *.egg-info
+per-file-ignores =
+    __init__.py:F401
+    tests/*:F401,F811

.github/workflows/test.yml

@@ -0,0 +1,120 @@
+name: Tests
+
+on:
+  push:
+    branches: [ master, develop ]
+  pull_request:
+    branches: [ master, develop ]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.10", "3.11", "3.12"]
+
+    services:
+      kinesis:
+        image: localstack/localstack:latest
+        ports:
+          - 4566:4566
+        env:
+          SERVICES: kinesis
+          KINESIS_ERROR_PROBABILITY: 0.0
+          DEBUG: 1
+
+      redis:
+        image: redis:latest
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v4
+      with:
+        python-version: ${{ matrix.python-version }}
+
+    - name: Cache pip dependencies
+      uses: actions/cache@v3
+      with:
+        path: ~/.cache/pip
+        key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt', '**/test-requirements.txt') }}
+        restore-keys: |
+          ${{ runner.os }}-pip-
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt
+        pip install -r test-requirements.txt
+        sudo apt-get update && sudo apt-get install -y netcat-openbsd redis-tools curl
+
+    - name: Wait for services
+      run: |
+        # Give services time to start
+        sleep 20
+        # Wait for LocalStack port to be available (skip health check due to S3 errors)
+        timeout 120 bash -c 'until nc -z localhost 4566; do echo "Waiting for localstack port..."; sleep 3; done'
+        echo "LocalStack port is ready"
+        # Wait for Redis
+        timeout 60 bash -c 'until redis-cli -h localhost ping; do echo "Waiting for redis..."; sleep 2; done'
+        echo "Redis is ready"
+
+    - name: Run tests
+      run: |
+        timeout 600 pytest -v --cov=kinesis --cov-report=xml --cov-report=term-missing
+      env:
+        ENDPOINT_URL: http://localhost:4566
+        REDIS_HOST: localhost
+        REDIS_PORT: 6379
+        AWS_DEFAULT_REGION: ap-southeast-2
+        AWS_ACCESS_KEY_ID: test
+        AWS_SECRET_ACCESS_KEY: test
+
+    - name: Upload coverage to Codecov
+      uses: codecov/codecov-action@v3
+      with:
+        file: ./coverage.xml
+        flags: unittests
+        name: codecov-umbrella
+        fail_ci_if_error: false
+
+  test-docker:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Run tests in Docker
+      run: |
+        docker compose up --build --abort-on-container-exit --exit-code-from test
+
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: "3.12"
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install black isort flake8
+
+    - name: Check code formatting with black
+      run: black --check --diff kinesis tests
+
+    - name: Check import sorting with isort
+      run: isort --check-only --diff kinesis tests
+
+    - name: Lint with flake8
+      run: flake8 kinesis tests --max-line-length=88 --extend-ignore=E203,W503

.gitignore

@@ -1,3 +1,4 @@
+venv/
 dist
 *.egg-info
 .idea
@@ -10,4 +11,5 @@ temp.py
 .mypy_cache
 .tox
 release.sh
-test_*
+test_*
+CLAUDE.md

Dockerfile

@@ -1,6 +1,6 @@
-FROM python:3.7-slim
+FROM python:3.12-slim
 
-RUN apt-get update && apt-get install -y gcc python-dev gettext-base
+RUN apt-get update && apt-get install -y gcc python3-dev gettext-base
 
 RUN mkdir /app
 
@@ -13,7 +13,6 @@ COPY test-requirements.txt /app/test-requirements.txt
 RUN pip install -r /app/test-requirements.txt
 
 COPY kinesis /app/kinesis/
-
-COPY tests.py /app/tests.py
+COPY tests /app/tests/
 
 WORKDIR /app/

docker-compose.yaml

@@ -1,8 +1,7 @@
-version: '3.3'
 services:
 
   kinesis:
-    image: vsouza/kinesis-local:latest
+    image: instructure/kinesalite:latest
     command: --port 4567 --shardLimit 10000
     restart: always
     ports:
@@ -15,10 +14,11 @@ services:
 
   test:
      container_name: async-kinesis-test
-     command: ['nosetests']
+     command: ['pytest', '-v', '--cov=kinesis']
      volumes:
-      - ./tests.py:/app/tests.py
+      - ./tests:/app/tests
       - ./kinesis:/app/kinesis
+      - ./pytest.ini:/app/pytest.ini
      build:
        context: .
        dockerfile: Dockerfile
@@ -27,8 +27,9 @@ services:
       - ENDPOINT_URL=http://kinesis:4567
       - REDIS_HOST=redis
       - REDIS_PORT=6379
-      - AWS_ACCESS_KEY_ID=
-      - AWS_SECRET_ACCESS_KEY=
-     links:
-      - kinesis:kinesis
-      - redis:redis
+      - AWS_ACCESS_KEY_ID=test
+      - AWS_SECRET_ACCESS_KEY=test
+     depends_on:
+      - kinesis
+      - redis
+

kinesis/base.py

@@ -1,11 +1,13 @@
 import asyncio
 import logging
-from async_timeout import timeout
+import time
+from typing import Optional, Any, Dict, List, Union
+
+from .timeout_compat import timeout
 from aiobotocore.session import AioSession
 from asyncio import CancelledError
 from botocore.exceptions import ClientError
 from botocore.config import Config
-import time
 
 from . import exceptions
 
@@ -15,37 +17,37 @@
 class Base:
     def __init__(
         self,
-        stream_name,
-        session=None,
-        endpoint_url=None,
-        region_name=None,
-        retry_limit=None,
-        expo_backoff=None,
-        expo_backoff_limit=120,
-        skip_describe_stream=False,
-        create_stream=False,
-        create_stream_shards=1,
-    ):
-
-        self.stream_name = stream_name
+        stream_name: str,
+        session: Optional[AioSession] = None,
+        endpoint_url: Optional[str] = None,
+        region_name: Optional[str] = None,
+        retry_limit: Optional[int] = None,
+        expo_backoff: Optional[float] = None,
+        expo_backoff_limit: int = 120,
+        skip_describe_stream: bool = False,
+        create_stream: bool = False,
+        create_stream_shards: int = 1,
+    ) -> None:
+
+        self.stream_name: str = stream_name
 
         if session:
             assert isinstance(session, AioSession)
-            self.session = session
+            self.session: AioSession = session
         else:
             self.session = AioSession()
 
-        self.endpoint_url = endpoint_url
-        self.region_name = region_name
+        self.endpoint_url: Optional[str] = endpoint_url
+        self.region_name: Optional[str] = region_name
 
-        self.client = None
-        self.shards = None
+        self.client: Optional[Any] = None  # aiobotocore client
+        self.shards: Optional[List[Dict[str, Any]]] = None
 
-        self.stream_status = None
+        self.stream_status: Optional[str] = None
 
-        self.retry_limit = retry_limit
-        self.expo_backoff = expo_backoff
-        self.expo_backoff_limit = expo_backoff_limit
+        self.retry_limit: Optional[int] = retry_limit
+        self.expo_backoff: Optional[float] = expo_backoff
+        self.expo_backoff_limit: int = expo_backoff_limit
 
         # connection states of kinesis client
         self.RECONNECT = "RECONNECT"
@@ -60,7 +62,7 @@ def __init__(
         self.create_stream = create_stream
         self.create_stream_shards = create_stream_shards
 
-    async def __aenter__(self):
+    async def __aenter__(self) -> "Base":
 
         log.info(
             "creating client with {}".format(
@@ -78,12 +80,12 @@ async def __aenter__(self):
 
         return self
 
-    async def __aexit__(self, exc_type, exc, tb):
+    async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
         await self.close()
         await self.client.__aexit__(exc_type, exc, tb)
 
     @property
-    def address(self):
+    def address(self) -> Dict[str, str]:
         """
         Return address of stream, either as StreamName or StreamARN, when applicable.
 

kinesis/checkpointers.py

@@ -3,20 +3,45 @@
 import os
 import json
 from datetime import timezone, datetime
+from typing import Protocol, Optional, Union, Dict, Tuple, Any
 
 log = logging.getLogger(__name__)
 
 
+class CheckPointer(Protocol):
+    """Protocol for checkpointer implementations."""
+    
+    async def allocate(self, shard_id: str) -> Tuple[bool, Optional[str]]:
+        """Allocate a shard for processing."""
+        ...
+    
+    async def deallocate(self, shard_id: str) -> None:
+        """Deallocate a shard."""
+        ...
+    
+    async def checkpoint(self, shard_id: str, sequence_number: str) -> None:
+        """Checkpoint progress for a shard."""
+        ...
+    
+    def get_all_checkpoints(self) -> Dict[str, str]:
+        """Get all checkpoints."""
+        ...
+    
+    async def close(self) -> None:
+        """Close the checkpointer."""
+        ...
+
+
 class BaseCheckPointer:
-    def __init__(self, name="", id=None):
-        self._id = id if id else os.getpid()
-        self._name = name
-        self._items = {}
+    def __init__(self, name: str = "", id: Optional[Union[str, int]] = None) -> None:
+        self._id: Union[str, int] = id if id else os.getpid()
+        self._name: str = name
+        self._items: Dict[str, Any] = {}
 
-    def get_id(self):
+    def get_id(self) -> Union[str, int]:
         return self._id
 
-    def get_ref(self):
+    def get_ref(self) -> str:
         return "{}/{}".format(self._name, self._id)
 
     def get_all_checkpoints(self):
@@ -84,6 +109,9 @@ def is_allocated(self, shard_id):
         return shard_id in self._items and self._items[shard_id]["active"]
 
     async def allocate(self, shard_id):
+        if self.is_allocated(shard_id):
+            return False, None
+            
         if shard_id not in self._items:
             self._items[shard_id] = {"sequence": None}
 
@@ -117,9 +145,9 @@ def __init__(
         )
 
         if is_cluster:
-            from aredis import StrictRedisCluster as Redis
+            from redis.asyncio.cluster import RedisCluster as Redis
         else:
-            from aredis import StrictRedis as Redis
+            from redis.asyncio import Redis
 
         params = {
             "host": os.environ.get("REDIS_HOST", "localhost"),