dinsaw/kines

Friendly CLI for Amazon Kinesis Data Streams

kines's Issues

Feature Request - Add in `Raw` decoded formatting for the `walk` subcommand

First off, thank you for putting this together! It's helpful to have something that masks over some of the UX and gets away from my hacky Bash polling loop!

One feature that would be appreciated is a --raw flag for the walk subcommand. Providing it, the tool would then only be piping out the base64 decoded values of the payload instead of the pretty formatting you have now.

This would make it simpler to copy/paste some of the payloads.

Read Aggregate Records

Issue #10 is occurring because kines does not support reading aggregated records yet.

binascii.Error: Incorrect padding

When running

kines walk stream-name 000000000001 -t '10m'

I got the following error:

Creating shard iterator with arguments = {'StreamName': 'stream-name', 'ShardId': 'shardId-000000000001', 'ShardIteratorType': 'AT_TIMESTAMP', 'Timestamp': datetime.datetime(2021, 1, 25, 15, 56, 10)}
No records found for this api call 😔
Fetch more records? [Y/n]: y
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/kines/read_util.py", line 85, in get_parsed_data
    return base64.b64decode(kinesis_record["Data"]).decode(UTF_8)
  File "/usr/local/Cellar/[email protected]/3.9.1_1/Frameworks/Python.framework/Versions/3.9/lib/python3.9/base64.py", line 87, in b64decode
    return binascii.a2b_base64(s)
binascii.Error: Incorrect padding

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/bin/kines", line 8, in <module>
    sys.exit(kines())
  File "/usr/local/lib/python3.9/site-packages/click/core.py", line 829, in __call__
    return self.main(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/click/core.py", line 782, in main
    rv = self.invoke(ctx)
  File "/usr/local/lib/python3.9/site-packages/click/core.py", line 1259, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.9/site-packages/click/core.py", line 1066, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/usr/local/lib/python3.9/site-packages/click/core.py", line 610, in invoke
    return callback(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/kines/kines_cli.py", line 100, in walk
    read_util.walk(
  File "/usr/local/lib/python3.9/site-packages/kines/read_util.py", line 56, in walk
    parsed_data = get_parsed_data(record)
  File "/usr/local/lib/python3.9/site-packages/kines/read_util.py", line 87, in get_parsed_data
    return kinesis_record["Data"].decode(UTF_8)
UnicodeDecodeError: 'utf-8' codec can't decode bytes in position 0-2: invalid continuation byte

[Feature] Other Kinesis shard iterator types

Hey,
it's a really useful tool!

What I am missing is the ability to debug stream in real time, needed during the development. There is a walk command, but it supports only two iterator types - TRIM_HORIZON (default) and AT_SEQUENCE_NUMBER, right?

Would be great to support also AT_TIMESTAMP, maybe be providing full datetime or time description like 10 minutes ago?

Also would be nice to have long-living polling - start the tool with LATEST iterator and just get all new incoming events (by calling Kinesis every second or two).

What do you think?

Add Put Record functionality

• Single Record
• Aggregate Record