firehose json record decorator

firehose_record_decorator

class Record:
    def __init__(self, record):
        self.record = record

    def decoded_data(self):
        return base64.b64decode(self.record['data'])

    def merge(self, new_data):
        return {
            'recordId': self.record['recordId'],
            'result': 'Ok',
            'data': self.encode_data(new_data)
        }

    def encode_data(self, new_data):
        return base64.encodebytes(new_data.encode()).decode()

class JsonRecord(Record):
    def decoded_data(self):
        return json.loads(super().decoded_data())

    def encode_data(self, new_data):
        data = json.dumps(new_data)
        return super().encode_data(data)

def firehose_json(func):
    def wrapper(record):
        jrecord = JsonRecord(record)
        raw_payload = func(jrecord.decoded_data())
        return jrecord.merge(raw_payload)
    return wrapper

@firehose_json
def transform(record):
    #do the fancy stuff here
    return record
   
def record_iterator(event, func):
    results = []
    for record in event['records']:
        transformed = func(record)
        if transformed:
            results.append(transformed)            
    return {'records': results}
        
def lambda_handler(event, context):
    return record_iterator(event, transform)