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

# vim: tabstop=4 shiftwidth=4 softtabstop=4

#    Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import abc

from taskflow import exceptions as exc
from taskflow import states
from taskflow import utils

from taskflow.openstack.common import uuidutils


def task_and_state(task, state):
    name_pieces = []
    try:
        name_pieces.append(task.name)
        if isinstance(task.version, (list, tuple)):
            name_pieces.append(utils.join(task.version, "."))
        else:
            name_pieces.append(task.version)
    except AttributeError:
        pass
    if not name_pieces:
        # Likely a function and not a task object so let us search for these
        # attributes to get a good name for this task.
        name_pieces = [a for a in utils.get_many_attr(task,
                                                      '__module__',
                                                      '__name__',
                                                      '__version__')
                       if a is not None]
    if not name_pieces:
        # Ok, unsure what this task is, just use whatever its string
        # representation is.
        name_pieces.append(task)
    return "%s;%s" % (utils.join(name_pieces, ':'), state)


class Claimer(object):
    """A base class for objects that can attempt to claim a given
    job, so that said job can be worked on."""

    __metaclass__ = abc.ABCMeta

    @abc.abstractmethod
    def claim(self, job, owner):
        """This method will attempt to claim said job and must
        either succeed at this or throw an exception signaling the job can not
        be claimed."""
        raise NotImplementedError()

    @abc.abstractmethod
    def unclaim(self, job, owner):
        """This method will attempt to unclaim said job and must
        either succeed at this or throw an exception signaling the job can not
        be unclaimed."""
        raise NotImplementedError()


class Job(object):
    """A job is connection to some set of work to be done by some agent. Basic
    information is provided about said work to be able to attempt to
    fullfill said work."""

    __metaclass__ = abc.ABCMeta

    def __init__(self, name, context, catalog, claimer, jid=None):
        self.name = name
        self.context = context
        self.owner = None
        self.posted_on = []
        self._catalog = catalog
        self._claimer = claimer
        self._logbook = None
        if not jid:
            self._id = uuidutils.generate_uuid()
        else:
            self._id = str(jid)
        self._state = states.UNCLAIMED

    def __str__(self):
        return "Job (%s, %s): %s" % (self.name, self.tracking_id, self.state)

    @property
    def state(self):
        return self._state

    @state.setter
    def state(self, new_state):
        self._change_state(new_state)

    def _change_state(self, new_state):
        if self.state != new_state:
            self._state = new_state
            # TODO(harlowja): add logbook info?

    def _workflow_listener(self, _context, flow, _old_state):
        """Ensure that when we receive an event from said workflow that we
        make sure a logbook entry exists for that flow."""
        if flow.name in self.logbook:
            return
        self.logbook.add_flow(flow.name)

    def _task_listener(self, _context, state, flow, task, result=None):
        """Store the result of the task under the given flow in the log
        book so that it can be retrieved later."""
        metadata = None
        flow_details = self.logbook[flow.name]
        if state == states.SUCCESS:
            metadata = {
                'result': result,
            }
        task_state = task_and_state(task, state)
        if task_state not in flow_details:
            task_details = flow_details.add_task(task_state)
            task_details.metadata = metadata

    def _task_result_fetcher(self, _context, flow, task):
        flow_details = self.logbook[flow.name]
        # See if it completed before so that we can use its results instead
        # of having to recompute them.
        task_state = task_and_state(task, states.SUCCESS)
        if task_state in flow_details:
            # TODO(harlowja): should we be a little more cautious about
            # duplicate task results? Maybe we shouldn't allow them to
            # have the same name in the first place?
            task_details = flow_details[task_state][0]
            if task_details.metadata and 'result' in task_details.metadata:
                return (True, task_details.metadata['result'])
        return (False, None)

    def associate(self, flow):
        """Attachs the needed resumption and state change tracking listeners
        to the given workflow so that the workflow can be resumed/tracked
        using the jobs components."""
        if self._task_listener not in flow.task_listeners:
            flow.task_listeners.append(self._task_listener)
        if self._workflow_listener not in flow.listeners:
            flow.listeners.append(self._workflow_listener)
        flow.result_fetcher = self._task_result_fetcher

    @property
    def logbook(self):
        """Fetches (or creates) a logbook entry for this job."""
        if self._logbook is None:
            self._logbook = self._catalog.create_or_fetch(self)
        return self._logbook

    def claim(self, owner):
        """This can be used to attempt transition this job from unclaimed
        to claimed.

        This must be done in a way that likely uses some type of locking or
        ownership transfer so that only a single entity gets this job to work
        on. This will avoid multi-job ownership, which can lead to
        inconsistent state."""
        if self.state != states.UNCLAIMED:
            raise exc.UnclaimableJobException("Unable to claim job when job is"
                                              " in state %s" % (self.state))
        self._claimer.claim(self, owner)
        self._change_state(states.CLAIMED)

    def run(self, flow, *args, **kwargs):
        already_associated = []

        def associate_all(a_flow):
            if a_flow in already_associated:
                return
            # Associate with the flow.<commit_msg>Add flow disassociation and adjust the assocate path<commit_after># -*- coding: utf-8 -*-

# vim: tabstop=4 shiftwidth=4 softtabstop=4

#    Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import abc

from taskflow import exceptions as exc
from taskflow import states
from taskflow import utils

from taskflow.openstack.common import uuidutils


def task_and_state(task, state):
    name_pieces = []
    try:
        name_pieces.append(task.name)
        if isinstance(task.version, (list, tuple)):
            name_pieces.append(utils.join(task.version, "."))
        else:
            name_pieces.append(task.version)
    except AttributeError:
        pass
    if not name_pieces:
        # Likely a function and not a task object so let us search for these
        # attributes to get a good name for this task.
        name_pieces = [a for a in utils.get_many_attr(task,
                                                      '__module__',
                                                      '__name__',
                                                      '__version__')
                       if a is not None]
    if not name_pieces:
        # Ok, unsure what this task is, just use whatever its string
        # representation is.
        name_pieces.append(task)
    return "%s;%s" % (utils.join(name_pieces, ':'), state)


class Claimer(object):
    """A base class for objects that can attempt to claim a given
    job, so that said job can be worked on."""

    __metaclass__ = abc.ABCMeta

    @abc.abstractmethod
    def claim(self, job, owner):
        """This method will attempt to claim said job and must
        either succeed at this or throw an exception signaling the job can not
        be claimed."""
        raise NotImplementedError()

    @abc.abstractmethod
    def unclaim(self, job, owner):
        """This method will attempt to unclaim said job and must
        either succeed at this or throw an exception signaling the job can not
        be unclaimed."""
        raise NotImplementedError()


class Job(object):
    """A job is connection to some set of work to be done by some agent. Basic
    information is provided about said work to be able to attempt to
    fullfill said work."""

    __metaclass__ = abc.ABCMeta

    def __init__(self, name, context, catalog, claimer, jid=None):
        self.name = name
        self.context = context
        self.owner = None
        self.posted_on = []
        self._catalog = catalog
        self._claimer = claimer
        self._logbook = None
        if not jid:
            self._id = uuidutils.generate_uuid()
        else:
            self._id = str(jid)
        self._state = states.UNCLAIMED

    def __str__(self):
        return "Job (%s, %s): %s" % (self.name, self.tracking_id, self.state)

    @property
    def state(self):
        return self._state

    @state.setter
    def state(self, new_state):
        self._change_state(new_state)

    def _change_state(self, new_state):
        if self.state != new_state:
            self._state = new_state
            # TODO(harlowja): add logbook info?

    def _workflow_listener(self, _context, flow, _old_state):
        """Ensure that when we receive an event from said workflow that we
        make sure a logbook entry exists for that flow."""
        if flow.name in self.logbook:
            return
        self.logbook.add_flow(flow.name)

    def _task_listener(self, _context, state, flow, task, result=None):
        """Store the result of the task under the given flow in the log
        book so that it can be retrieved later."""
        metadata = None
        flow_details = self.logbook[flow.name]
        if state == states.SUCCESS:
            metadata = {
                'result': result,
            }
        task_state = task_and_state(task, state)
        if task_state not in flow_details:
            task_details = flow_details.add_task(task_state)
            task_details.metadata = metadata

    def _task_result_fetcher(self, _context, flow, task):
        flow_details = self.logbook[flow.name]
        # See if it completed before so that we can use its results instead
        # of having to recompute them.
        task_state = task_and_state(task, states.SUCCESS)
        if task_state in flow_details:
            # TODO(harlowja): should we be a little more cautious about
            # duplicate task results? Maybe we shouldn't allow them to
            # have the same name in the first place?
            task_details = flow_details[task_state][0]
            if task_details.metadata and 'result' in task_details.metadata:
                return (True, task_details.metadata['result'])
        return (False, None)

    def associate(self, flow, parents=True):
        """Attachs the needed resumption and state change tracking listeners
        to the given workflow so that the workflow can be resumed/tracked
        using the jobs components."""

        if self._task_listener not in flow.task_listeners:
            flow.task_listeners.append(self._task_listener)
        if self._workflow_listener not in flow.listeners:
            flow.listeners.append(self._workflow_listener)
        flow.result_fetcher = self._task_result_fetcher

        # Associate the parents as well (if desired)
        if parents and flow.parents:
            for p in flow.parents:
                self.associate(p, parents)

    def disassociate(self, flow, parents=True):
        """Detaches the needed resumption and state change tracking listeners
        from the given workflow."""

        if self._task_listener in flow.task_listeners:
            flow.task_listeners.remove(self._task_listener)
        if self._workflow_listener in flow.listeners:
            flow.listeners.remove(self._workflow_listener)
        if flow.result_fetcher is self._task_result_fetcher:
            flow.result_fetcher = None

        # Disassociate from the flows parents (if desired)
        if parents and flow.parents:
            for p in flow.parents:
                self.disassociate(p, parents)

    @property
    def logbook(self):
        """Fetches (or creates) a logbook entry for this job."""
        if self._logbook is None:
            self._logbook = self._catalog.create_or_fetch(self)
        return self._logbook

    def claim(self, owner):
        """This can be used to attempt transition this job from unclaimed
        to claimed.

        This must be done in a way that likely uses some type of locking or
        ownership transfer so that only a single entity gets this job to work
        on. This will avoid multi-job ownership, which can lead to
        inconsistent state."""
        if self.state != states.UNCLAIMED:
            raise exc.UnclaimableJobException("Unable to claim job when job is"
                                              " in state %s" % (self.state))
        self._claimer.claim(self, owner)
        self._change_state(states.CLAIMED)

    def run(self, flow, *args, **kwargs):
        already_associated = []

        def associate_all(a_flow):
            if a_flow in already_associated:
                return
            # Associate with the flow.
{'fraction_non_alphanumeric': 0.0744135148074687, 'fraction_numerical': 0.0017782641406196566, 'mean_word_length': 2.195367132867133, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4576961', 'n_tokens_mistral': 3928, 'n_tokens_neox': 3811, 'n_words': 1438}