values

augment_values(values, schemas, constants=None)

Source code in kiara/utils/values.py

def augment_values(
    values: Mapping[str, Any],
    schemas: Mapping[str, ValueSchema],
    constants: Optional[Mapping[str, ValueSchema]] = None,
) -> Dict[str, Any]:

    # TODO: check if extra fields were provided

    if constants:
        for k, v in constants.items():
            if k in values.keys():
                raise Exception(f"Invalid input: value provided for constant '{k}'")

    values_new = {}

    if constants:
        for field_name, schema in constants.items():
            v = schema.default
            assert v not in [None, SpecialValue.NO_VALUE, SpecialValue.NOT_SET]
            if callable(v):
                values_new[field_name] = v()
            else:
                values_new[field_name] = copy.deepcopy(v)

    for field_name, schema in schemas.items():

        if field_name in values_new.keys():
            raise Exception(
                f"Duplicate field '{field_name}', this is most likely a bug."
            )

        if field_name not in values.keys():
            if schema.default != SpecialValue.NOT_SET:
                if callable(schema.default):
                    values_new[field_name] = schema.default()
                else:
                    values_new[field_name] = copy.deepcopy(schema.default)
            else:
                values_new[field_name] = SpecialValue.NOT_SET
        else:
            value = values[field_name]
            if value is None:
                value = SpecialValue.NO_VALUE
            values_new[field_name] = value

    return values_new

create_schema_dict(schema_config)

Source code in kiara/utils/values.py

def create_schema_dict(
    schema_config: Mapping[str, Union[ValueSchema, Mapping[str, Any]]],
) -> Mapping[str, ValueSchema]:

    invalid = check_valid_field_names(*schema_config.keys())
    if invalid:
        raise Exception(
            f"Can't assemble schema because it contains invalid input field name(s) '{', '.join(invalid)}'. Change the input schema to not contain any of the reserved keywords: {', '.join(INVALID_VALUE_NAMES)}"
        )

    result = {}
    for k, v in schema_config.items():

        if isinstance(v, ValueSchema):
            result[k] = v
        elif isinstance(v, Mapping):
            _v = dict(v)
            if "doc" not in _v.keys():
                _v["doc"] = DEFAULT_NO_DESC_VALUE
            schema = ValueSchema(**_v)

            result[k] = schema
        else:
            if v is None:
                msg = "None"
            else:
                msg = v.__class__
            raise Exception(
                f"Invalid return type '{msg}' for field '{k}' when trying to create schema."
            )

    return result

overlay_constants_and_defaults(schemas, defaults, constants)

Source code in kiara/utils/values.py

def overlay_constants_and_defaults(
    schemas: Mapping[str, ValueSchema],
    defaults: Mapping[str, Any],
    constants: Mapping[str, Any],
):

    for k, v in schemas.items():

        default_value = defaults.get(k, None)
        constant_value = constants.get(k, None)

        # value_to_test = None
        if default_value is not None and constant_value is not None:
            raise Exception(
                f"Module configuration error. Value '{k}' set in both 'constants' and 'defaults', this is not allowed."
            )

        # TODO: perform validation for constants/defaults

        if default_value is not None:
            schemas[k].default = default_value

        if constant_value is not None:
            schemas[k].default = constant_value
            schemas[k].is_constant = True

    input_schemas = {}
    constants = {}
    for k, v in schemas.items():
        if v.is_constant:
            constants[k] = v
        else:
            input_schemas[k] = v

    return input_schemas, constants