<commit_before># -*- coding:utf-8 -*-

# Copyright (c) 2013, <NAME>
# Copyright (c) 2013, <NAME>
#
# See the file LICENSE for copying permission.

import time
import json
import collections

from boto.swf.exceptions import SWFResponseError, SWFTypeAlreadyExistsError

from swf.constants import REGISTERED
from swf.utils import immutable
from swf.models import BaseModel, Domain
from swf.models.history import History
from swf.models.base import ModelDiff
from swf import exceptions
from swf.exceptions import (
    DoesNotExistError,
    AlreadyExistsError,
    ResponseError,
    raises,
)


_POLICIES = (
    'TERMINATE',       # child executions will be terminated
    'REQUEST_CANCEL',  # a request to cancel will be attempted for
                       # each child execution
    'ABANDON',         # no action will be taken
)

CHILD_POLICIES = collections.namedtuple('CHILD_POLICY',
                                        ' '.join(_POLICIES))(*_POLICIES)


class WorkflowTypeDoesNotExist(DoesNotExistError):
    pass


class WorkflowExecutionDoesNotExist(DoesNotExistError):
    pass


@immutable
class WorkflowType(BaseModel):
    """Simple Workflow Type wrapper

    :param  domain: Domain the workflow type should be registered in
    :type   domain: swf.models.Domain

    :param  name: name of the workflow type
    :type   name: String

    :param  version: workflow type version
    :type   version: String

    :param  status: workflow type status
    :type   status: swf.core.ConnectedSWFObject.{REGISTERED, DEPRECATED}

    :param   creation_date: creation date of the current WorkflowType
    :type    creation_date: float (timestamp)

    :param   deprecation_date: deprecation date of WorkflowType
    :type    deprecation_date: float (timestamp)

    :param  task_list: task list to use for scheduling decision tasks for executions
                       of this workflow type
    :type   task_list: String

    :param  child_policy: policy to use for the child workflow executions
                          when a workflow execution of this type is terminated
    :type   child_policy: CHILD_POLICIES.{TERMINATE |
                                          REQUEST_CANCEL |
                                          ABANDON}

    :param  execution_timeout: maximum duration for executions of this workflow type
    :type   execution_timeout: String

    :param  decision_tasks_timeout: maximum duration of decision tasks for this workflow type
    :type   decision_tasks_timeout: String

    :param  description: Textual description of the workflow type
    :type   description: String
    """
    __slots__ = [
        'domain',
        'name',
        'version',
        'status',
        'creation_date',
        'deprecation_date',
        'task_list',
        'child_policy',
        'execution_timeout',
        'decision_tasks_timeout',
        'description',
    ]

    def __init__(self, domain, name, version,
                 status=REGISTERED,
                 creation_date=0.0,
                 deprecation_date=0.0,
                 task_list=None,
                 child_policy=CHILD_POLICIES.TERMINATE,
                 execution_timeout='300',
                 decision_tasks_timeout='300',
                 description=None, *args, **kwargs):
        self.domain = domain
        self.name = name
        self.version = version
        self.status = status
        self.creation_date = creation_date
        self.deprecation_date = deprecation_date

        self.task_list = task_list
        self.execution_timeout = execution_timeout
        self.decision_tasks_timeout = decision_tasks_timeout
        self.description = description

        # Explicitly call child_policy setter
        # to validate input value
        self.set_child_policy(child_policy)

        # immutable decorator rebinds class name,
        # so have to use generice self.__class__
        super(self.__class__, self).__init__(*args, **kwargs)

    def set_child_policy(self, policy):
        if not policy in CHILD_POLICIES:
            raise ValueError("invalid child policy value: {}".format(policy))

        self.child_policy = policy

    def _diff(self):
        """Checks for differences between WorkflowType instance
        and upstream version

        :returns: A list of swf.models.base.ModelDiff namedtuple describing
                  differences
        :rtype: list
        """
        try:
            description = self.connection.describe_workflow_type(
                self.domain.name,
                self.name,
                self.version
            )
        except SWFResponseError as e:
            if e.error_code == 'UnknownResourceFault':
                raise DoesNotExistError("Remote Domain does not exist")

            raise ResponseError(e.body['message'])

        workflow_info = description['typeInfo']
        workflow_config = description['configuration']

        return ModelDiff(
            ('name', self.name, workflow_info['workflowType']['name']),
            ('version', self.version, workflow_info['workflowType']['version']),
            ('status', self.status, workflow_info['status']),
            ('creation_date', self.creation_date, workflow_info['creationDate']),
            ('deprecation_date', self.deprecation_date, workflow_info['deprecationDate']),
            ('task_list', self.task_list, workflow_config['defaultTaskList']['name']),
            ('child_policy', self.child_policy, workflow_config['defaultChildPolicy']),
            ('execution_timeout', self.execution_timeout, workflow_config['defaultExecutionStartToCloseTimeout']),
            ('decision_tasks_timout', self.decision_tasks_timeout, workflow_config['defaultTaskStartToCloseTimeout']),
            ('description', self.description, workflow_info['description']),
        )

    @property
    @exceptions.translate(SWFResponseError, to=ResponseError)
    @exceptions.is_not(WorkflowTypeDoesNotExist)
    @exceptions.when(SWFResponseError,
                     raises(WorkflowTypeDoesNotExist,
                            when=exceptions.is_unknown('WorkflowType'),
                            extract=exceptions.extract_resource))
    def exists(self):
        """Checks if the WorkflowType exists amazon-side

        :rtype: bool
        """
        self.connection.describe_workflow_type(
            self.domain.name,
            self.name,
            self.version
        )
        return True

    def save(self):
        """Creates the workflow type amazon side"""
        try:
            self.connection.register_workflow_type(
                self.domain.name,
                self.name,
                self.version,
                task_list=str(self.task_list),
                default_child_policy=str(self.child_policy),
                default_execution_start_to_close_timeout=str(self.execution_timeout),
                default_task_start_to_close_timeout=str(self.decision_tasks_timeout),
                description=self.description
            )
        except SWFTypeAlreadyExistsError:
            raise AlreadyExistsError("Workflow type %s already exists amazon-side" % self.name)
        except SWFResponseError as e:
            if e.error_code == 'UnknownResourceFault':
                raise DoesNotExistError(e.body['message'])

    def delete(self):
        """Deprecates the workflow type amazon-side"""
        try:
            self.connection.deprecate_workflow_type(self.domain.name, self.name, self.version)
        except SWFResponseError as e:
            if e.error_code in ['UnknownResourceFault', 'TypeDeprecatedFault']:
                raise DoesNotExistError(e.body['message'])

    def upstream(self):
        from swf.querysets.workflow import WorkflowTypeQuerySet
        qs = WorkflowTypeQuerySet(self.domain)
        return qs.get(self.name, self.version)

    def start_execution(self, workflow_id=None, task_list=None,
                        child_policy=None, execution_timeout=None,
                        input=None, tag_list=None, decision_tasks_timeout=None):
        """Starts a Workflow execution of current workflow type

        :param  workflow_id: The user defined identifier associated with the workflow execution
        :type   workflow_id: String

        :param  task_list: task list to use for scheduling decision tasks for execution
                        of this workflow
        :type   task_list: String

        :param  child_policy: policy to use for the child workflow executions
                              of this workflow execution.
        :type   child_policy: CHILD_POLICIES.{TERMINATE |
                                              REQUEST_CANCEL |
                                              ABANDON}

        :param  execution_timeout: maximum duration for the workflow execution
        :type   execution_timeout: String

        :param  input: Input of the workflow execution
        :type   input: dict

        :param  tag_list: Tags associated with the workflow execution
        :type   tag_list: String or list of strings

        :param  decision_tasks_timeout: maximum duration of decision tasks
                                        for this workflow execution
        :type   decision_tasks_timeout: String
        """
        workflow_id = workflow_id or '%s-%s-%i' % (self.name, self.version, time.time())
        task_list = task_list or self.task_list
        child_policy = child_policy or self.child_policy
        input = json.dumps(input) or None
        tag_list = tag_list if isinstance(tag_list, list) else [tag_list]

        run_id = self.connection.start_workflow_execution(<commit_msg>Raise an error if too many tags when starting an execution (#113)<commit_after># -*- coding:utf-8 -*-

# Copyright (c) 2013, <NAME>
# Copyright (c) 2013, <NAME>
#
# See the file LICENSE for copying permission.

import time
import json
import collections

from boto.swf.exceptions import SWFResponseError, SWFTypeAlreadyExistsError

from swf.constants import REGISTERED
from swf.utils import immutable
from swf.models import BaseModel, Domain
from swf.models.history import History
from swf.models.base import ModelDiff
from swf import exceptions
from swf.exceptions import (
    DoesNotExistError,
    AlreadyExistsError,
    ResponseError,
    raises,
)


_POLICIES = (
    'TERMINATE',       # child executions will be terminated
    'REQUEST_CANCEL',  # a request to cancel will be attempted for
                       # each child execution
    'ABANDON',         # no action will be taken
)

CHILD_POLICIES = collections.namedtuple('CHILD_POLICY',
                                        ' '.join(_POLICIES))(*_POLICIES)


class WorkflowTypeDoesNotExist(DoesNotExistError):
    pass


class WorkflowExecutionDoesNotExist(DoesNotExistError):
    pass


@immutable
class WorkflowType(BaseModel):
    """Simple Workflow Type wrapper

    :param  domain: Domain the workflow type should be registered in
    :type   domain: swf.models.Domain

    :param  name: name of the workflow type
    :type   name: String

    :param  version: workflow type version
    :type   version: String

    :param  status: workflow type status
    :type   status: swf.core.ConnectedSWFObject.{REGISTERED, DEPRECATED}

    :param   creation_date: creation date of the current WorkflowType
    :type    creation_date: float (timestamp)

    :param   deprecation_date: deprecation date of WorkflowType
    :type    deprecation_date: float (timestamp)

    :param  task_list: task list to use for scheduling decision tasks for executions
                       of this workflow type
    :type   task_list: String

    :param  child_policy: policy to use for the child workflow executions
                          when a workflow execution of this type is terminated
    :type   child_policy: CHILD_POLICIES.{TERMINATE |
                                          REQUEST_CANCEL |
                                          ABANDON}

    :param  execution_timeout: maximum duration for executions of this workflow type
    :type   execution_timeout: String

    :param  decision_tasks_timeout: maximum duration of decision tasks for this workflow type
    :type   decision_tasks_timeout: String

    :param  description: Textual description of the workflow type
    :type   description: String
    """
    __slots__ = [
        'domain',
        'name',
        'version',
        'status',
        'creation_date',
        'deprecation_date',
        'task_list',
        'child_policy',
        'execution_timeout',
        'decision_tasks_timeout',
        'description',
    ]

    def __init__(self, domain, name, version,
                 status=REGISTERED,
                 creation_date=0.0,
                 deprecation_date=0.0,
                 task_list=None,
                 child_policy=CHILD_POLICIES.TERMINATE,
                 execution_timeout='300',
                 decision_tasks_timeout='300',
                 description=None, *args, **kwargs):
        self.domain = domain
        self.name = name
        self.version = version
        self.status = status
        self.creation_date = creation_date
        self.deprecation_date = deprecation_date

        self.task_list = task_list
        self.execution_timeout = execution_timeout
        self.decision_tasks_timeout = decision_tasks_timeout
        self.description = description

        # Explicitly call child_policy setter
        # to validate input value
        self.set_child_policy(child_policy)

        # immutable decorator rebinds class name,
        # so have to use generice self.__class__
        super(self.__class__, self).__init__(*args, **kwargs)

    def set_child_policy(self, policy):
        if not policy in CHILD_POLICIES:
            raise ValueError("invalid child policy value: {}".format(policy))

        self.child_policy = policy

    def _diff(self):
        """Checks for differences between WorkflowType instance
        and upstream version

        :returns: A list of swf.models.base.ModelDiff namedtuple describing
                  differences
        :rtype: list
        """
        try:
            description = self.connection.describe_workflow_type(
                self.domain.name,
                self.name,
                self.version
            )
        except SWFResponseError as e:
            if e.error_code == 'UnknownResourceFault':
                raise DoesNotExistError("Remote Domain does not exist")

            raise ResponseError(e.body['message'])

        workflow_info = description['typeInfo']
        workflow_config = description['configuration']

        return ModelDiff(
            ('name', self.name, workflow_info['workflowType']['name']),
            ('version', self.version, workflow_info['workflowType']['version']),
            ('status', self.status, workflow_info['status']),
            ('creation_date', self.creation_date, workflow_info['creationDate']),
            ('deprecation_date', self.deprecation_date, workflow_info['deprecationDate']),
            ('task_list', self.task_list, workflow_config['defaultTaskList']['name']),
            ('child_policy', self.child_policy, workflow_config['defaultChildPolicy']),
            ('execution_timeout', self.execution_timeout, workflow_config['defaultExecutionStartToCloseTimeout']),
            ('decision_tasks_timout', self.decision_tasks_timeout, workflow_config['defaultTaskStartToCloseTimeout']),
            ('description', self.description, workflow_info['description']),
        )

    @property
    @exceptions.translate(SWFResponseError, to=ResponseError)
    @exceptions.is_not(WorkflowTypeDoesNotExist)
    @exceptions.when(SWFResponseError,
                     raises(WorkflowTypeDoesNotExist,
                            when=exceptions.is_unknown('WorkflowType'),
                            extract=exceptions.extract_resource))
    def exists(self):
        """Checks if the WorkflowType exists amazon-side

        :rtype: bool
        """
        self.connection.describe_workflow_type(
            self.domain.name,
            self.name,
            self.version
        )
        return True

    def save(self):
        """Creates the workflow type amazon side"""
        try:
            self.connection.register_workflow_type(
                self.domain.name,
                self.name,
                self.version,
                task_list=str(self.task_list),
                default_child_policy=str(self.child_policy),
                default_execution_start_to_close_timeout=str(self.execution_timeout),
                default_task_start_to_close_timeout=str(self.decision_tasks_timeout),
                description=self.description
            )
        except SWFTypeAlreadyExistsError:
            raise AlreadyExistsError("Workflow type %s already exists amazon-side" % self.name)
        except SWFResponseError as e:
            if e.error_code == 'UnknownResourceFault':
                raise DoesNotExistError(e.body['message'])

    def delete(self):
        """Deprecates the workflow type amazon-side"""
        try:
            self.connection.deprecate_workflow_type(self.domain.name, self.name, self.version)
        except SWFResponseError as e:
            if e.error_code in ['UnknownResourceFault', 'TypeDeprecatedFault']:
                raise DoesNotExistError(e.body['message'])

    def upstream(self):
        from swf.querysets.workflow import WorkflowTypeQuerySet
        qs = WorkflowTypeQuerySet(self.domain)
        return qs.get(self.name, self.version)

    def start_execution(self, workflow_id=None, task_list=None,
                        child_policy=None, execution_timeout=None,
                        input=None, tag_list=None, decision_tasks_timeout=None):
        """Starts a Workflow execution of current workflow type

        :param  workflow_id: The user defined identifier associated with the workflow execution
        :type   workflow_id: String

        :param  task_list: task list to use for scheduling decision tasks for execution
                        of this workflow
        :type   task_list: String

        :param  child_policy: policy to use for the child workflow executions
                              of this workflow execution.
        :type   child_policy: CHILD_POLICIES.{TERMINATE |
                                              REQUEST_CANCEL |
                                              ABANDON}

        :param  execution_timeout: maximum duration for the workflow execution
        :type   execution_timeout: String

        :param  input: Input of the workflow execution
        :type   input: dict

        :param  tag_list: Tags associated with the workflow execution
        :type   tag_list: String or list of strings

        :param  decision_tasks_timeout: maximum duration of decision tasks
                                        for this workflow execution
        :type   decision_tasks_timeout: String
        """
        workflow_id = workflow_id or '%s-%s-%i' % (self.name, self.version, time.time())
        task_list = task_list or self.task_list
        child_policy = child_policy or self.child_policy
        input = json.dumps(input) or None
        tag_list = tag_list if isinstance(tag_list, list) else [tag_list]

        # checks
        if len(tag_list) > 5:
            raise ValueError("You cannot have more than 5 tags in StartWorkflowExecution.")

        run_id = self.connection.start_workflow_execution(
{'fraction_non_alphanumeric': 0.07570476678626345, 'fraction_numerical': 0.002203997949769349, 'mean_word_length': 2.2935516542876435, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 26, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2846141', 'n_tokens_mistral': 5257, 'n_tokens_neox': 4981, 'n_words': 1321}