A python utility for testing lambda kinesis handlers locally using real data.
You want to test your lambda handler locally on real data before deploying it. This utility reads from a kinesis stream, transforms the events and forwards them to your handler.
At the time of writing, AWS SAM supports running kinesis handlers locally on a custom event.
pip install local_lambda_kinesis
lambda_kinesis -s <stream_name> -p <handler path>
-i <iterator type>
- Set the shard iterator type, supporsLATEST
andTRIM_HORIZON
.--shard-iterator
- By default the handler runs on the first available shard in the stream. you can use a specific one.
The tool uses the default AWS CLI settings. To override them, set AWS_PROFILE
and AWS_DEFAULT_REGION
.
Testing is done using , and linting is done using and [mypy]( pytest --disable-warnings @mypy --ignore-missing-imports . @black --check -l 100 . @flake8 . --max-line-length=100
Three main tools are used for development:
- pytest
- flake8 - for basic linting.
- black - format the code to enforce style.
- mypy - for type checking.
Contribution is always welcome. To contibute:
- Fork the repository
- Create a pull request
- Make sure the tests pass
if you're having issues with black, run
make format-code
orblack -l 100 .
.