<commit_before>from django.core.exceptions import ObjectDoesNotExist
from analysis.exceptions import *
from analysis.models.base import AnalysisAppInstanceModel, AnalysisAppImmutableModel
from analysis.models.data_objects import DataObject
from analysis.models.task_definitions import TaskDefinition
from analysis.models.task_runs import TaskRun, TaskRunInput, TaskRunOutput
from jinja2 import DictLoader, Environment
from universalmodels import fields


"""
This module defines WorkflowRun and other classes related to
receiving and running a request for analysis.
"""


class WorkflowRun(AnalysisAppInstanceModel):
    """WorkflowRun represents a request to execute a Workflow on a particular
    set of inputs, and the execution status of that Workflow
    """

    NAME_FIELD = 'workflow__workflow_name'

    workflow = fields.ForeignKey('Workflow')
    workflow_run_inputs = fields.OneToManyField('WorkflowRunInput')
    workflow_run_outputs = fields.OneToManyField('WorkflowRunOutput')
    step_runs = fields.OneToManyField('StepRun', related_name='workflow_run')
    channels = fields.OneToManyField('Channel')
    status = fields.CharField(
        max_length=255,
        default='running',
        choices=(('running', 'Running'),
                 ('canceled', 'Canceled'),
                 ('completed', 'Completed')
        )
    )

    @classmethod
    def update_status_for_all(cls):
        for workflow_run in cls.objects.filter(status='running'):
            workflow_run.update_status()

    def update_status(self):
        for step_run in self.step_runs.filter(status='waiting') | self.step_runs.filter(status='running'):
            step_run.update_status()
        step_run_statuses = [step_run.status for step_run in self.step_runs.all()]
        if all([status == 'completed' for status in step_run_statuses]):
            self.update({'status': 'completed'})
        if any([status == 'error' for status in step_run_statuses]):
            self.update({'status': 'error'})

    def cancel(self):
        for step_run in self.step_runs.filter(status='waiting') | self.step_runs.filter(status='running'):
            step_run.cancel()
        self.update({'status': 'canceled'})
    
    @classmethod
    def order_by_most_recent(cls, count=None):
        workflow_runs = cls.objects.order_by('datetime_created').reverse()
        if count is not None and (workflow_runs.count() > count):
            return workflow_runs[:count]
        else:
            return workflow_runs

    @classmethod
    def create(cls, *args, **kwargs):
        workflow_run = super(WorkflowRun, cls).create(*args, **kwargs)
        # WorkflowRunInputs must be already created by the client
        workflow_run._sync_outputs()
        workflow_run._sync_step_runs()
        workflow_run._sync_channels()
        # TODO Assign workflow status (probably w/ default field)
        return workflow_run

    def _sync_outputs(self):
        for workflow_output in self.workflow.workflow_outputs.all():
            self._sync_output(workflow_output)

    def _sync_output(self, workflow_output):
        """Create a workflow_run_output corresponding each workflow_output if it does not already exist.
        """
        try:
            self.workflow_run_outputs.get(workflow_output___id=workflow_output._id)
            return
        except ObjectDoesNotExist:
            self.workflow_run_outputs.add(
                WorkflowRunOutput.create(
                    {'workflow_output': workflow_output.to_struct()}
                )
            )

    def _sync_step_runs(self):
        for step in self.workflow.steps.all():
            self._sync_step_run(step)

    def _sync_step_run(self, step):
        """Create a step_run corresponding each step if it does not already exist.
        """
        try:
            self.step_runs.get(step___id=step._id)<commit_msg>validate channel names on new workflows<commit_after>from django.core.exceptions import ObjectDoesNotExist
from analysis.exceptions import *
from analysis.models.base import AnalysisAppInstanceModel, AnalysisAppImmutableModel
from analysis.models.data_objects import DataObject
from analysis.models.task_definitions import TaskDefinition
from analysis.models.task_runs import TaskRun, TaskRunInput, TaskRunOutput
from jinja2 import DictLoader, Environment
from universalmodels import fields


"""
This module defines WorkflowRun and other classes related to
receiving and running a request for analysis.
"""


class WorkflowRun(AnalysisAppInstanceModel):
    """WorkflowRun represents a request to execute a Workflow on a particular
    set of inputs, and the execution status of that Workflow
    """

    NAME_FIELD = 'workflow__workflow_name'

    workflow = fields.ForeignKey('Workflow')
    workflow_run_inputs = fields.OneToManyField('WorkflowRunInput')
    workflow_run_outputs = fields.OneToManyField('WorkflowRunOutput')
    step_runs = fields.OneToManyField('StepRun', related_name='workflow_run')
    channels = fields.OneToManyField('Channel')
    status = fields.CharField(
        max_length=255,
        default='running',
        choices=(('running', 'Running'),
                 ('canceled', 'Canceled'),
                 ('completed', 'Completed')
        )
    )

    @classmethod
    def update_status_for_all(cls):
        for workflow_run in cls.objects.filter(status='running'):
            workflow_run.update_status()

    def update_status(self):
        for step_run in self.step_runs.filter(status='waiting') | self.step_runs.filter(status='running'):
            step_run.update_status()
        step_run_statuses = [step_run.status for step_run in self.step_runs.all()]
        if all([status == 'completed' for status in step_run_statuses]):
            self.update({'status': 'completed'})
        if any([status == 'error' for status in step_run_statuses]):
            self.update({'status': 'error'})

    def cancel(self):
        for step_run in self.step_runs.filter(status='waiting') | self.step_runs.filter(status='running'):
            step_run.cancel()
        self.update({'status': 'canceled'})
    
    @classmethod
    def order_by_most_recent(cls, count=None):
        workflow_runs = cls.objects.order_by('datetime_created').reverse()
        if count is not None and (workflow_runs.count() > count):
            return workflow_runs[:count]
        else:
            return workflow_runs

    @classmethod
    def create(cls, *args, **kwargs):
        workflow_run = super(WorkflowRun, cls).create(*args, **kwargs)
        # WorkflowRunInputs must be already created by the client
        workflow_run._sync_outputs()
        workflow_run._sync_step_runs()
        workflow_run._sync_channels()
        # TODO Assign workflow status (probably w/ default field)
        return workflow_run

    @classmethod
    def validate_create_input(cls, workflow_run_struct):
        # Structure will be validated by creating object.
        # However, channels are implicitly defined, and we have to validate to avoid
        # confusing errors creating channels.

        # First get a list of to_channel names and make sure there are no duplicates
        def add_input(channel_inputs, channel_name):
            if channel_name in channel_inputs:
                raise Exception("The channel '%s' has two data sources. Only one is allowed." % channel_name)
            channel_inputs.append(channel_name)
            
        channel_inputs = []
        if workflow_run_struct.get('workflow_run_inputs') is not None:
            for workflow_run_input in workflow_run_struct['workflow_run_inputs']:
                if workflow_run_input.get('workflow_input') is None:
                    raise Exception('workflow_run_input definition is missing a workflow_input. %s' % workflow_run_input)
                if workflow_run_input['workflow_input'].get('to_channel') is None:
                    raise Exception('workflow_input is missing channel. %s' % workflow_run_input['workflow_input'])
                channel_name = workflow_run_input['workflow_input']['to_channel']
                add_input(channel_inputs, channel_name)
        if workflow_run_struct.get('workflow') is None:
            raise Exception('WorkflowRun definition is invalid. Missing workflow. %s' % workflow_run_struct)
        if workflow_run_struct['workflow']['steps'] is None:
            raise Exception('Workflow definition is invalid. Missing steps. %s' % workflow_run_struct['workflow'])
        for step in workflow_run_struct['workflow']['steps']:
            if step.get('step_outputs') is None:
                raise Exception('Step definition is invalid, missing step_outputs. %s' % step)
            for step_output in step['step_outputs']:
                if step_output.get('to_channel') is None:
                    raise Exception('step_output definition is invalid. Missing "to_channel". %s' % step_output)
                channel_name = step_output['to_channel']
                add_input(channel_inputs, channel_name)

        # Now verify that every from_channel has a source channel defined
        def check_source(channel_inputs, channel_name):
            if channel_name not in channel_inputs:
                raise Exception("from_channel '%s' is defined, but it has no source. "\
                                "There should be a corresponding workflow_input or a step_output with to_channel '%s'"
                                % (channel_name, channel_name))

        if workflow_run_struct['workflow'].get('workflow_outputs') is None:
            raise Exception('Invalid workflow, missing workflow_outputs. %s' % workflow_run_struct['workflow'])
        for workflow_output in  workflow_run_struct['workflow']['workflow_outputs']:
            if workflow_output.get('from_channel') is None:
                raise Exception('Invalid workflow output, from_channel is missing. %s' % workflow_output)
            channel_name = workflow_output['from_channel']
            check_source(channel_inputs, channel_name)
        for step in workflow_run_struct['workflow']['steps']:
            if step.get('step_inputs') is not None:
                for step_input in step['step_inputs']:
                    if step_input.get('from_channel') is None:
                        raise Exception('step_input definition is invalid. Missing "from_channel". %s' % step_input)
                    channel_name = step_input.get('from_channel')
                    check_source(channel_inputs, channel_name)
                
    def _sync_outputs(self):
        for workflow_output in self.workflow.workflow_outputs.all():
            self._sync_output(workflow_output)

    def _sync_output(self, workflow_output):
        """Create a workflow_run_output corresponding each workflow_output if it does not already exist.
        """
        try:
            self.workflow_run_outputs.get(workflow_output___id=workflow_output._id)
            return
        except ObjectDoesNotExist:
            self.workflow_run_outputs.add(
                WorkflowRunOutput.create(
                    {'workflow_output': workflow_output.to_struct()}
                )
            )

    def _sync_step_runs(self):
        for step in self.workflow.steps.all():
            self._sync_step_run(step)

    def _sync_step_run(self, step):
        """Create a step_run corresponding each step if it does not already exist.
        """
        try:
            self.step_runs.get(step___id=step._id)
{'fraction_non_alphanumeric': 0.08908821736480015, 'fraction_numerical': 0.0006966820517286423, 'mean_word_length': 2.903467029231815, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2002139', 'n_tokens_mistral': 3193, 'n_tokens_neox': 3076, 'n_words': 774}