def _set_constants()

in metaflow/flowspec.py [0:0]

• 52 lines of code
• 22 McCabe index (conditional complexity)

    def _set_constants(self, graph, kwargs, config_options):
        from metaflow.decorators import (
            flow_decorators,
        )  # To prevent circular dependency

        # Persist values for parameters and other constants (class level variables)
        # only once. This method is called before persist_constants is called to
        # persist all values set using setattr
        self._check_parameters(config_parameters=False)

        seen = set()
        self._success = True

        parameters_info = []
        for var, param in self._get_parameters():
            seen.add(var)
            if param.IS_CONFIG_PARAMETER:
                # Use computed value if already evaluated, else get from config_options
                val = param._computed_value or config_options.get(
                    param.name.replace("-", "_").lower()
                )
            else:
                val = kwargs[param.name.replace("-", "_").lower()]
            # Support for delayed evaluation of parameters.
            if isinstance(val, DelayedEvaluationParameter):
                val = val()
            val = val.split(param.separator) if val and param.separator else val
            if isinstance(val, ConfigValue):
                # We store config values as dict so they are accessible with older
                # metaflow clients. It also makes it easier to access.
                val = val.to_dict()
            setattr(self, var, val)
            parameters_info.append({"name": var, "type": param.__class__.__name__})

        # Do the same for class variables which will be forced constant as modifications
        # to them don't propagate well since we create a new process for each step and
        # re-read the flow file
        constants_info = []
        for var in dir(self.__class__):
            if var[0] == "_" or var in self._NON_PARAMETERS or var in seen:
                continue
            val = getattr(self.__class__, var)
            if isinstance(val, (MethodType, FunctionType, property, type)):
                continue
            constants_info.append({"name": var, "type": type(val).__name__})
            setattr(self, var, val)

        # We store the DAG information as an artifact called _graph_info
        steps_info, graph_structure = graph.output_steps()

        graph_info = {
            "file": os.path.basename(os.path.abspath(sys.argv[0])),
            "parameters": parameters_info,
            "constants": constants_info,
            "steps": steps_info,
            "graph_structure": graph_structure,
            "doc": graph.doc,
            "decorators": [
                {
                    "name": deco.name,
                    "attributes": to_pod(deco.attributes),
                    "statically_defined": deco.statically_defined,
                }
                for deco in flow_decorators(self)
                if not deco.name.startswith("_")
            ],
            "extensions": extension_info(),
        }
        self._graph_info = graph_info