def verify_challenge_response()

in source/backend/chalicelib/framework.py [0:0]

• 60 lines of code
• 10 McCabe index (conditional complexity)

def verify_challenge_response(challenge_id):
    blueprint.log.debug('verify_challenge_response: %s', challenge_id)
    # Looking up challenge on DynamoDB table
    item = _table.get_item(Key={'id': challenge_id})
    if 'Item' not in item:
        blueprint.log.info('Challenge not found: %s', challenge_id)
        raise NotFoundError('Challenge not found')
    challenge = _read_item(item['Item'])
    blueprint.log.debug('challenge: %s', challenge)
    # Getting challenge type, params and frames
    challenge_type = challenge['type']
    params = challenge['params']
    frames = challenge['frames']
    # Invoking Rekognition with parallel threads
    with ThreadPoolExecutor(max_workers=_THREAD_POOL_SIZE) as pool:
        futures = [
            pool.submit(
                _detect_faces, frame
            ) for frame in frames
        ]
        frames = [r.result() for r in as_completed(futures)]
    frames.sort(key=lambda frame: frame['key'])
    current_state = _challenge_state_funcs[challenge_type][_FIRST_STATE]
    context = dict()
    end_times = dict()
    success = False
    for frame in frames:
        try:
            while True:
                blueprint.log.debug('----------------')
                blueprint.log.debug('current_state: %s', current_state.__name__)
                blueprint.log.debug('frame[timestamp]: %s', frame['timestamp'])
                blueprint.log.debug('context.keys: %s', context.keys())
                blueprint.log.debug('end_times: %s', end_times)
                next_state = current_state(params, frame, context, end_times)
                if next_state.__name__ != current_state.__name__:
                    current_state = next_state
                    blueprint.log.debug('NEXT')
                else:
                    blueprint.log.debug('CONTINUE')
                    break
        except _Success:
            success = True
            break
        except _Fail:
            break
    # Returning result based on final state
    blueprint.log.debug('success: %s', success)
    response = {'success': success}
    blueprint.log.debug('response: %s', response)
    # Updating challenge on DynamoDB table
    _table.update_item(
        Key={'id': challenge_id},
        UpdateExpression='set #frames = :frames, #success = :success',
        ExpressionAttributeNames={
            '#frames': 'frames',
            '#success': 'success'
        },
        ExpressionAttributeValues={
            ':frames': _write_item(frames),
            ':success': response['success']
        },
        ReturnValues='NONE'
    )
    return response