def __init__()

in liminal/runners/airflow/operators/operator_with_variable_resolving.py [0:0]

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

    def __init__(self, variables, task_config=None):
        super().__init__()
        self.val = None
        self.variables = variables.copy()
        logging.info(f'variables: {variables}')
        if task_config and 'variables' in task_config:
            task_variables = task_config['variables']
            if isinstance(task_variables, dict):
                self.variables.update(task_variables)
            elif isinstance(task_variables, str):
                variables_key = self.from_string(task_variables).render()
                if variables_key in variables:
                    self.variables.update(variables[variables_key])