@beam.ptransform_fn
@beam.typehints.with_input_types(types.Extracts)
@beam.typehints.with_output_types(evaluator.Evaluation)
def ExtractAndEvaluate(extracts: beam.pvalue.PCollection, extractors: List[extractor.Extractor], evaluators: List[evaluator.Evaluator]):
    'Performs Extractions and Evaluations in provided order.'
    evaluation = {}

    def update(evaluation: Dict[(Text, Any)], new_evaluation: Dict[(Text, Any)]):
        for (k, v) in new_evaluation.items():
            if (k not in evaluation):
                evaluation[k] = []
            evaluation[k].append(v)
        return evaluation
    for v in evaluators:
        if (not v.run_after):
            update(evaluation, (extracts | (v.stage_name >> v.ptransform)))
    for x in extractors:
        extracts = (extracts | (x.stage_name >> x.ptransform))
        for v in evaluators:
            if (v.run_after == x.stage_name):
                update(evaluation, (extracts | (v.stage_name >> v.ptransform)))
    for v in evaluators:
        if (v.run_after == extractor.LAST_EXTRACTOR_STAGE_NAME):
            update(evaluation, (extracts | (v.stage_name >> v.ptransform)))
    result = {}
    for (k, v) in evaluation.items():
        if (len(v) == 1):
            result[k] = v[0]
            continue
        result[k] = ((v | (('FlattenEvaluationOutput(%s)' % k) >> beam.Flatten())) | (('CombineEvaluationOutput(%s)' % k) >> beam.CombinePerKey(_CombineEvaluationDictionariesFn())))
    return result
Performs Extractions and Evaluations in provided order.
@beam.ptransform_fn
@beam.typehints.with_input_types(types.Extracts)
@beam.typehints.with_output_types(evaluator.Evaluation)
def ExtractAndEvaluate(extracts: beam.pvalue.PCollection, extractors: List[extractor.Extractor], evaluators: List[evaluator.Evaluator]):
    
    evaluation = {}

    def update(evaluation: Dict[(Text, Any)], new_evaluation: Dict[(Text, Any)]):
        for (k, v) in new_evaluation.items():
            if (k not in evaluation):
                evaluation[k] = []
            evaluation[k].append(v)
        return evaluation
    for v in evaluators:
        if (not v.run_after):
            update(evaluation, (extracts | (v.stage_name >> v.ptransform)))
    for x in extractors:
        extracts = (extracts | (x.stage_name >> x.ptransform))
        for v in evaluators:
            if (v.run_after == x.stage_name):
                update(evaluation, (extracts | (v.stage_name >> v.ptransform)))
    for v in evaluators:
        if (v.run_after == extractor.LAST_EXTRACTOR_STAGE_NAME):
            update(evaluation, (extracts | (v.stage_name >> v.ptransform)))
    result = {}
    for (k, v) in evaluation.items():
        if (len(v) == 1):
            result[k] = v[0]
            continue
        result[k] = ((v | (('FlattenEvaluationOutput(%s)' % k) >> beam.Flatten())) | (('CombineEvaluationOutput(%s)' % k) >> beam.CombinePerKey(_CombineEvaluationDictionariesFn())))
    return result