def resolve()

in statefun-sdk-python/statefun/storage.py [0:0]

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

def resolve(storage: StorageSpec,
            typename: str,
            values: typing.List[ToFunction.PersistedValue]) -> Resolution:
    """
    Resolve the registered specs and the actually received values.

    :param storage: a storage factory
    :param typename: the typename of the function under invocation
    :param values: the actually received values
    :return: a Resolution result, that might have either a list of missing specs
    (specs that were defined by the user but didn't arrived from StateFun) or a
     successful resolution, with an instance of an addressed scoped storage.
    """
    #
    # index the actually received values (TypedValue) by name.
    #
    received: typing.Dict[str, TypedValue] = {state.state_name: state.state_value for state in values}
    #
    # see if any of the specs are missing.
    #
    missing_keys = storage.names - received.keys()
    if missing_keys:
        # keep the missing specs in exactly the same order as they were originally defined.
        # This is not strictly required from the protocol point of view, but it makes it a bit easier
        # to troubleshoot.
        missing = [spec for spec in storage.specs if spec.name in missing_keys]
        return Resolution(missing_specs=missing, storage=None)
    else:
        cells: typing.Dict[str, Cell] = {spec.name: Cell(tpe=spec.type, typed_value=received[spec.name]) for spec in
                                         storage.specs}
        s = storage.make_instance(cells, typename)
        return Resolution(missing_specs=None, storage=s)