def unpack_kinesis_event(kinesis_event, deserializer=None, unpacker=None,<EOL>embed_timestamp=False):
records = kinesis_event["<STR_LIT>"]<EOL>events = []<EOL>shard_ids = set()<EOL>for rec in records:<EOL><INDENT>data = rec["<STR_LIT>"]["<STR_LIT:data>"]<EOL>try:<EOL><INDENT>payload = b64decode(data)<EOL><DEDENT>except TypeError:<EOL><INDENT>payload = b64decode(data.encode("<STR_LIT:utf-8>"))<EOL><DEDENT>if unpacker:<EOL><INDENT>payload = unpacker(payload)<EOL><DEDENT>shard_ids.add(rec["<STR_LIT>"].split("<STR_LIT::>")[<NUM_LIT:0>])<EOL>try:<EOL><INDENT>payload = payload.decode()<EOL><DEDENT>except AttributeError:<EOL><INDENT>pass<EOL><DEDENT>if deserializer:<EOL><INDENT>try:<EOL><INDENT>payload = deserializer(payload)<EOL><DEDENT>except ValueError:<EOL><INDENT>try:<EOL><INDENT>payload = deserializer(payload.replace("<STR_LIT>", "<STR_LIT:'>"))<EOL><DEDENT>except:<EOL><INDENT>logger.error("<STR_LIT>".format(<EOL>payload))<EOL>raise<EOL><DEDENT><DEDENT><DEDENT>if isinstance(payload, dict) and embed_timestamp:<EOL><INDENT>ts = rec["<STR_LIT>"].get("<STR_LIT>")<EOL>if ts:<EOL><INDENT>ts = datetime.fromtimestamp(ts, tz=tz.tzutc())<EOL>ts_str = ("<STR_LIT>"<EOL>"<STR_LIT>").format(<EOL>year=ts.year,<EOL>month=ts.month,<EOL>day=ts.day,<EOL>hour=ts.hour,<EOL>minute=ts.minute,<EOL>second=ts.second)<EOL><DEDENT>else:<EOL><INDENT>ts_str = "<STR_LIT>"<EOL><DEDENT>payload[embed_timestamp] = ts_str<EOL><DEDENT>events.append(payload)<EOL><DEDENT>if len(shard_ids) > <NUM_LIT:1>:<EOL><INDENT>msg = "<STR_LIT>".format(<EOL>shard_ids)<EOL>raise(BadKinesisEventError(msg))<EOL><DEDENT>return events, shard_ids.pop()<EOL>
Extracts events (a list of dicts) from a Kinesis event.