Repository: ansible/awx
File: awx/main/tasks/jobs.py (Python)

Code before:
        Hook for any steps to run before the job/task starts
        """
        instance.log_lifecycle("pre_run")

        # Before task is started, ensure that job_event partitions exist
        create_partition(instance.event_class._meta.db_table, start=instance.created)

    def post_run_hook(self, instance, status):
        """
        Hook for any steps to run before job/task is marked as complete.
        """
        instance.log_lifecycle("post_run")

    def final_run_hook(self, instance, status, private_data_dir):
        """
        Hook for any steps to run after job/task is marked as complete.
        """
        instance.log_lifecycle("finalize_run")
        artifact_dir = os.path.join(private_data_dir, 'artifacts', str(self.instance.id))
        data_file_path = os.path.join(artifact_dir, 'ansible_data.json')

        if os.path.exists(data_file_path):
            with open(data_file_path) as f:
                collected_data = json.loads(f.read())

            instance.installed_collections = collected_data['installed_collections']
            instance.ansible_version = collected_data['ansible_version']
            instance.save(update_fields=['installed_collections', 'ansible_version'])

        # Run task manager appropriately for speculative dependencies
        if instance.unifiedjob_blocked_jobs.exists():
            ScheduleTaskManager().schedule()
        if instance.spawned_by_workflow:
            ScheduleWorkflowManager().schedule()

    def should_use_fact_cache(self):
        return False

    @with_path_cleanup
    @with_signal_handling
    def run(self, pk, **kwargs):
        """
        Run the job/task and capture its output.
        """
        self.instance = self.model.objects.get(pk=pk)
        if self.instance.status != 'canceled' and self.instance.cancel_flag:
            self.instance = self.update_model(self.instance.pk, start_args='', status='canceled')
        if self.instance.status not in ACTIVE_STATES:
            # Prevent starting the job if it has been reaped or handled by another process.
            raise RuntimeError(f'Not starting {self.instance.status} task pk={pk} because {self.instance.status} is not a valid active state')


Review comment (suggestion):
This could throw a `ValueError` if the plugin returns invalid or out-of-spec json.

Code after:

    def pre_run_hook(self, instance, private_data_dir):
        """
        Hook for any steps to run before the job/task starts
        """
        instance.log_lifecycle("pre_run")

        # Before task is started, ensure that job_event partitions exist
        create_partition(instance.event_class._meta.db_table, start=instance.created)

    def post_run_hook(self, instance, status):
        """
        Hook for any steps to run before job/task is marked as complete.
        """
        instance.log_lifecycle("post_run")

    def final_run_hook(self, instance, status, private_data_dir):
        """
        Hook for any steps to run after job/task is marked as complete.
        """
        instance.log_lifecycle("finalize_run")
        if flag_enabled("FEATURE_INDIRECT_NODE_COUNTING_ENABLED"):
            artifact_dir = os.path.join(private_data_dir, 'artifacts', str(self.instance.id))
            data_file_path = os.path.join(artifact_dir, 'ansible_data.json')

            if os.path.exists(data_file_path):
                with open(data_file_path) as f:
                    collected_data = json.loads(f.read())

                instance.installed_collections = collected_data['installed_collections']
                instance.ansible_version = collected_data['ansible_version']
                instance.save(update_fields=['installed_collections', 'ansible_version'])

        # Run task manager appropriately for speculative dependencies
        if instance.unifiedjob_blocked_jobs.exists():
            ScheduleTaskManager().schedule()
        if instance.spawned_by_workflow:
            ScheduleWorkflowManager().schedule()

    def should_use_fact_cache(self):
        return False

    @with_path_cleanup
    @with_signal_handling
    def run(self, pk, **kwargs):
        """
        Run the job/task and capture its output.
        """
        self.instance = self.model.objects.get(pk=pk)
        if self.instance.status != 'canceled' and self.instance.cancel_flag:
            self.instance = self.update_model(self.instance.pk, start_args='', status='canceled')