def lambda_handler()

in source/mediainfo/lambda_function.py [0:0]

• 40 lines of code
• 7 McCabe index (conditional complexity)

def lambda_handler(event, context):
    print(f'REQUEST:: {json.dumps(event)}')

    try:
        metadata = {}
        metadata['filename'] = event['srcVideo']

        dir_path = os.path.dirname(os.path.realpath(__file__))
        executable_path = os.path.join(dir_path, 'bin', 'mediainfo')

        signed_url = get_signed_url(event['srcBucket'], event['srcVideo'])
        json_content = json.loads(subprocess.check_output([executable_path, '--Output=JSON', signed_url]))
        print(f'MEDIAINFO OUTPUT:: {json.dumps(json_content)}')

        tracks = json_content['media']['track']
        for track in tracks:
            track_type = track['@type']
            if (track_type == 'General'):
                metadata['container'] = parse_general_attributes(track)
            elif (track_type == 'Video'):
                metadata.setdefault('video', []).append(parse_video_attributes(track))
            elif (track_type == 'Audio'):
                metadata.setdefault('audio', []).append(parse_audio_attributes(track))
            elif (track_type == 'Text'):
                metadata.setdefault('text', []).append(parse_text_attributes(track))
            else:
                print(f'Unsupported: {track_type}')

        event['srcMediainfo'] = json.dumps(metadata, indent=2)
        print(f'RESPONSE:: {json.dumps(metadata)}')

        return event
    except Exception as err:
        payload = {
            'guid': event['guid'],
            'event': event,
            'function': os.environ.get('AWS_LAMBDA_FUNCTION_NAME'),
            'error': str(err)
        }

        lambda_cli = boto3.client('lambda')
        lambda_cli.invoke(
            FunctionName=os.environ.get('ErrorHandler'),
            InvocationType='RequestResponse',
            Payload=json.dumps(payload)
        )

        raise err