https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
airflow_context_vars = context_to_airflow_vars(context, in_env_var_format=True)
            os.environ.update(airflow_context_vars)
            
            
            
            if not self.next_method:
                self.log.info(
                    "Exporting env vars: %s",
                    " ".join(f"{k}={v!r}" for k, v in airflow_context_vars.items()),
                )
            
            
            self.task.pre_execute(context=context)  
            
            self._run_execute_callback(context, self.task)
            
            get_listener_manager().hook.on_task_instance_running(
                previous_state=TaskInstanceState.QUEUED, task_instance=self, session=session
            )
            
            with set_current_context(context):
                result = self._execute_task(context, task_orig)
            
            
            self.task.post_execute(context=context, result=result)  
        Stats.incr(f"operator_successes_{self.task.task_type}", tags=self.stats_tags)
        
        Stats.incr("operator_successes", tags={**self.stats_tags, "task_type": self.task.task_type})
        Stats.incr("ti_successes", tags=self.stats_tags)
    def _execute_task(self, context, task_orig):