Skip to content

Kinesis Configurations Details

The kinesis event will automatically handle many common things done when eventing off a kinesis stream. Developers then have the ability to further extend that functionality with custom middleware. Below is a full list of all the configurations available and examples of their use.

Examples

Don't like reading documentation? Then look at our examples, which can be deployed in 1 command into your AWS account! 🤓

Lambda Configuration

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
functions:
    kinesis-handler:
        handler: service/handlers/kinesis.handle
        memorySize: 512
        timeout: 30
        events:
            - stream:
                type: kinesis
                arn:
                Fn::GetAtt: [ MyKinesisStream, Arn ]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from acai_aws.kinesis.requirements import requirements
from acai_aws.common import logger

@requirements(
    before=log_something,
    data_class=SomeClass,
    raise_operation_error=True,
    raise_body_error=True,
    schema='service/openapi.yml',
    required_body='v1-kinesis-body', # or jsonschema dict
    after=alert_something,
)
def handle(event):
    for record in event.records:
        logger.log(log=record)

# example data class
class SomeClass:
    def __init__(self, record):
        for k, v in record.body.items():
            setattr(self, k, v)

# example before function
def log_something(records, requirements):
    if 'something' in requirements:
        logger.log(log=records) 

# example after function
def alert_something(records, result, requirements):
    if 'something' in result and 'alert' in requirements:
        logger.log(log=records)

Requirements Configuration Options

option type required default description
before func no None a custom function to be ran before your records are pulled
after func no None a custom function to be ran after your records are pulled
data_class class no None a custom class that will be passed instead of the records object
raise_body_error bool no False will raise exception if body of record does not match schema provided
required_body str or dict no None will validate body of record against this schema
schema str no None file path pointing to the location of the openapi.yml file

Kinesis Record Properties

property type description
id str id of the stream
name str name of the stream
source_arn str source arn of the stream
region str region arn of the stream
version str version arn of the stream
invoke_identity_arn str arn of the indentity
schema_version str schema version
partition_key str partition key
time_stamp str time stamp
sequence_number str sequence number
data any can be anything, automaticallyed b64 decoded objects
body any can be anything, automaticallyed b64 decoded objects

record.id

1
2
3
4
print(record.id);

# output
'shardId-000000000006:49590338271490256608559692538361571095921575989136588898'

record.name

1
2
3
4
print(record.name);

# output
'aws:kinesis:record'

record.source_arn

1
2
3
4
print(record.source_arn);

# output
'arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream'

record.region

1
2
3
4
print(record.region);

# output
'us-east-2'

record.version

1
2
3
4
print(record.version);

# output
'1.0'

record.invoke_identity_arn

1
2
3
4
print(record.invoke_identity_arn);

# output
'arn:aws:iam::123456789012:role/lambda-role'

record.schema_version

1
2
3
4
print(record.schema_version);

# output
'1.0'

record.partition_key

1
2
3
4
print(record.partition_key);

# output
'1'

record.time_stamp

1
2
3
4
print(record.time_stamp);

# output
1545084650.987

record.sequence_number

1
2
3
4
print(record.sequence_number);

# output
49590338271490256608559692538361571095921575989136588898

record.data

1
2
3
4
5
6
print(record.data);

# output
{
    'new_data': '123456789'
}

record.body

1
2
3
4
5
6
print(record.body);

# output
{
    'new_data': '123456789'
}