Skip to content
High level python library for using kinesis streams
Branch: master
Clone or download
Latest commit e944541 Dec 15, 2017
Type Name Latest commit message Commit time
Failed to load latest commit information.
pynesis Raise custom exception on read instead of boto3 ones Oct 24, 2017
.dockerignore Improve build speed by not sharing .tox from host and adding a named … May 18, 2017
.editorconfig Begin extracting kinesis related functionality from service into its … May 17, 2017
.flake8 Begin extracting kinesis related functionality from service into its … May 17, 2017
.gitignore Ignore build output directory Jun 8, 2017
.pylintrc Setup pylint to also use 120 characters lines Oct 28, 2017
AUTHORS Add authors Jun 8, 2017 Move breaking changes list to its own file Jun 12, 2017
Dockerfile Fix build & bump mypy Dec 15, 2017
LICENSE.txt Add license Jun 8, 2017
Makefile Add publish Makefile target for uploading releases to pypi Jun 13, 2017 Add pypi version badge and python-supported-versions badge Dec 15, 2017
circle.yml Begin extracting kinesis related functionality from service into its … May 17, 2017 Add django checkpointer May 24, 2017
setup.cfg Adding classifiers for pypi Dec 15, 2017

pynesis CircleCI PyPI version Python Versions

High level kinesis client. Support python 2.7 and 3.6, and has helpers for using it within Django.

Some features:

  • Supports python 2 & 3
  • Django helpers included
  • Automatically detects shard count changes
  • Checkpoints/sequences persistence can be customized
  • Provided Checkpointer implementations for memory, django model and redis
  • Provided Dummy kinesis implementation for development/testing

Some limitations:

  • Single threaded/sequential. It will read from all shards in a round-robin fashion


Install the latest stable version from Pypi with pip install pynesis


from pynesis.streams import KinesisStream

# Get a reference to the stream we want to work with
stream = KinesisStream("my-stream", region_name="eu-west-2")

# Write to the stream
stream.put("key", "my message".encode("utf-8"))

# Read from the stream
for record in

Now persisting the sequences:

from pynesis.streams import KinesisStream
from pynesis.checkpointers import RedisCheckpointer

checkpointer = RedisCheckpointer(redis_host="localhost")
stream = KinesisStream("my-stream", region_name="eu-west-2", checkpointer=checkpointer)

for record in
    print(record) returns a KinesisRecord on each iteration which has the following instance attributes for accessing the details of the raw record:

  • sequence_number
  • approximate_arrival_timestamp
  • data
  • partition_key

See the examples available here for more details

Django support

If you are using django, you can configure your kinesis streams in the standard django file, see here for an example.

You can use also the provided Django model based Checkpointer if you want to save streams sequences into the database instead of redis, for using it, just add pynesis to INSTALLED_APPS and run the provided migration with migrate.

Development environment

Run make shell and then use the tests/examples.

Running tests

To run all tests in all environments and python versions supported, run:

make test

To run a single test in a single environment, from within a make shell run:

tox -e py36-dj19-redis -- pynesis/tests/
You can’t perform that action at this time.