<commit_before># -*- coding: utf-8 -*-
##############################################################################
#
#    Author: <NAME>
#    Copyright 2013 Camptocamp SA
#
#    This program is free software: you can redistribute it and/or modify
#    it under the terms of the GNU Affero General Public License as
#    published by the Free Software Foundation, either version 3 of the
#    License, or (at your option) any later version.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU Affero General Public License for more details.
#
#    You should have received a copy of the GNU Affero General Public License
#    along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
##############################################################################

import sys
import logging
import inspect
import uuid
from datetime import datetime, timedelta, MINYEAR
from pickle import loads, dumps, UnpicklingError

from openerp import SUPERUSER_ID
from openerp.tools import DEFAULT_SERVER_DATETIME_FORMAT
from openerp.tools.translate import _

from ..exception import (NotReadableJobError,
                         NoSuchJobError,
                         FailedJobError,
                         RetryableJobError)

PENDING = 'pending'
ENQUEUED = 'enqueued'
DONE = 'done'
STARTED = 'started'
FAILED = 'failed'

STATES = [(PENDING, 'Pending'),
          (ENQUEUED, 'Enqueued'),
          (STARTED, 'Started'),
          (DONE, 'Done'),
          (FAILED, 'Failed')]

DEFAULT_PRIORITY = 10  # used by the PriorityQueue to sort the jobs
DEFAULT_MAX_RETRIES = 5
RETRY_INTERVAL = 10 * 60  # seconds

_logger = logging.getLogger(__name__)


def _unpickle(pickled):
    """ Unpickles a string and catch all types of errors it can throw,
    to raise only NotReadableJobError in case of error.

    OpenERP stores the text fields as 'utf-8', so we specify the encoding.

    `loads()` may raises many types of exceptions (AttributeError,
    IndexError, TypeError, KeyError, ...). They are all catched and
    raised as `NotReadableJobError`).
    """
    try:
        unpickled = loads(pickled.encode('utf-8'))
    except (StandardError, UnpicklingError):
        raise NotReadableJobError('Could not unpickle.', pickled)
    return unpickled


class JobStorage(object):
    """ Interface for the storage of jobs """

    def store(self, job):
        """ Store a job """
        raise NotImplementedError

    def load(self, job_uuid):
        """ Read the job's data from the storage """
        raise NotImplementedError

    def exists(self, job_uuid):
        """Returns if a job still exists in the storage."""
        raise NotImplementedError


class OpenERPJobStorage(JobStorage):
    """ Store a job on OpenERP """

    _job_model_name = 'queue.job'
    _worker_model_name = 'queue.worker'

    def __init__(self, session):
        super(OpenERPJobStorage, self).__init__()
        self.session = session
        self.job_model = self.session.pool.get(self._job_model_name)
        self.worker_model = self.session.pool.get(self._worker_model_name)
        assert self.job_model is not None, (
            "Model %s not found" % self._job_model_name)

    def enqueue(self, func, model_name=None, args=None, kwargs=None,
                priority=None, eta=None, max_retries=None, description=None):
        """Create a Job and enqueue it in the queue. Return the job uuid.

        This expects the arguments specific to the job to be already extracted
        from the ones to pass to the job function.

        """
        job = Job(func=func, model_name=model_name, args=args, kwargs=kwargs,
                  priority=priority, eta=eta, max_retries=max_retries, description=description)
        job.user_id = self.session.uid
        self.store(job)
        return job.uuid

    def enqueue_resolve_args(self, func, *args, **kwargs):
        """Create a Job and enqueue it in the queue. Return the job uuid."""
        priority = kwargs.pop('priority', None)
        eta = kwargs.pop('eta', None)
        model_name = kwargs.pop('model_name', None)
        max_retries = kwargs.pop('max_retries', None)
        description = kwargs.pop('description', None)

        return self.enqueue(func, model_name=model_name,
                            args=args, kwargs=kwargs,
                            priority=priority,
                            max_retries=max_retries,
                            eta=eta,
                            description=description)

    def exists(self, job_uuid):
        """Returns if a job still exists in the storage."""
        return bool(self._openerp_id(job_uuid))

    def _openerp_id(self, job_uuid):
        openerp_id = None
        job_ids = self.job_model.search(self.session.cr,
                                        SUPERUSER_ID,
                                        [('uuid', '=', job_uuid)],
                                        context=self.session.context,
                                        limit=1)
        if job_ids:
            openerp_id = job_ids[0]
        return openerp_id

    def openerp_id(self, job):
        return self._openerp_id(job.uuid)

    def _worker_id(self, worker_uuid):
        openerp_id = None
        worker_ids = self.worker_model.search(self.session.cr,
                                              SUPERUSER_ID,
                                              [('uuid', '=', worker_uuid)],
                                              context=self.session.context,
                                              limit=1)
        if worker_ids:
            openerp_id = worker_ids[0]
        return openerp_id

    def store(self, job):
        """ Store the Job """
        vals = {'state': job.state,
                'priority': job.priority,
                'retry': job.retry,
                'max_retries': job.max_retries,
                'exc_info': job.exc_info,
                'user_id': job.user_id or self.session.uid,
                'result': unicode(job.result) if job.result else False,
                'date_enqueued': False,
                'date_started': False,
                'date_done': False,
                'eta': False,
                }

        if job.date_enqueued:
            vals['date_enqueued'] = job.date_enqueued.strftime(
                DEFAULT_SERVER_DATETIME_FORMAT)
        if job.date_started:
            vals['date_started'] = job.date_started.strftime(
                DEFAULT_SERVER_DATETIME_FORMAT)
        if job.date_done:
            vals['date_done'] = job.date_done.strftime(
                DEFAULT_SERVER_DATETIME_FORMAT)
        if job.eta:
            vals['eta'] = job.eta.strftime(DEFAULT_SERVER_DATETIME_FORMAT)

        if job.canceled:
            vals['active'] = False

        if job.worker_uuid:
            vals['worker_id'] = self._worker_id(job.worker_uuid)
        else:
            vals['worker_id'] = False

        if self.exists(job.uuid):
            self.job_model.write(self.session.cr,
                                 self.session.uid,
                                 self.openerp_id(job),
                                 vals,
                                 self.session.context)
        else:
            date_created = job.date_created.strftime(DEFAULT_SERVER_DATETIME_FORMAT)
            vals.update({'uuid': job.uuid,
                         'name': job.description,
                         'func_string': job.func_string,
                         'date_created': date_created,
                         'model_name': (job.model_name if job.model_name
                                        else False),
                         })

            vals['func'] = dumps((job.func_name,
                                  job.args,
                                  job.kwargs))

            self.job_model.create(self.session.cr,
                                  self.session.uid,
                                  vals,<commit_msg>The creation of a job is low level and should not be restrained by the accesses of the user.<commit_after># -*- coding: utf-8 -*-
##############################################################################
#
#    Author: <NAME>
#    Copyright 2013 Camptocamp SA
#
#    This program is free software: you can redistribute it and/or modify
#    it under the terms of the GNU Affero General Public License as
#    published by the Free Software Foundation, either version 3 of the
#    License, or (at your option) any later version.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU Affero General Public License for more details.
#
#    You should have received a copy of the GNU Affero General Public License
#    along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
##############################################################################

import sys
import logging
import inspect
import uuid
from datetime import datetime, timedelta, MINYEAR
from pickle import loads, dumps, UnpicklingError

from openerp import SUPERUSER_ID
from openerp.tools import DEFAULT_SERVER_DATETIME_FORMAT
from openerp.tools.translate import _

from ..exception import (NotReadableJobError,
                         NoSuchJobError,
                         FailedJobError,
                         RetryableJobError)

PENDING = 'pending'
ENQUEUED = 'enqueued'
DONE = 'done'
STARTED = 'started'
FAILED = 'failed'

STATES = [(PENDING, 'Pending'),
          (ENQUEUED, 'Enqueued'),
          (STARTED, 'Started'),
          (DONE, 'Done'),
          (FAILED, 'Failed')]

DEFAULT_PRIORITY = 10  # used by the PriorityQueue to sort the jobs
DEFAULT_MAX_RETRIES = 5
RETRY_INTERVAL = 10 * 60  # seconds

_logger = logging.getLogger(__name__)


def _unpickle(pickled):
    """ Unpickles a string and catch all types of errors it can throw,
    to raise only NotReadableJobError in case of error.

    OpenERP stores the text fields as 'utf-8', so we specify the encoding.

    `loads()` may raises many types of exceptions (AttributeError,
    IndexError, TypeError, KeyError, ...). They are all catched and
    raised as `NotReadableJobError`).
    """
    try:
        unpickled = loads(pickled.encode('utf-8'))
    except (StandardError, UnpicklingError):
        raise NotReadableJobError('Could not unpickle.', pickled)
    return unpickled


class JobStorage(object):
    """ Interface for the storage of jobs """

    def store(self, job):
        """ Store a job """
        raise NotImplementedError

    def load(self, job_uuid):
        """ Read the job's data from the storage """
        raise NotImplementedError

    def exists(self, job_uuid):
        """Returns if a job still exists in the storage."""
        raise NotImplementedError


class OpenERPJobStorage(JobStorage):
    """ Store a job on OpenERP """

    _job_model_name = 'queue.job'
    _worker_model_name = 'queue.worker'

    def __init__(self, session):
        super(OpenERPJobStorage, self).__init__()
        self.session = session
        self.job_model = self.session.pool.get(self._job_model_name)
        self.worker_model = self.session.pool.get(self._worker_model_name)
        assert self.job_model is not None, (
            "Model %s not found" % self._job_model_name)

    def enqueue(self, func, model_name=None, args=None, kwargs=None,
                priority=None, eta=None, max_retries=None, description=None):
        """Create a Job and enqueue it in the queue. Return the job uuid.

        This expects the arguments specific to the job to be already extracted
        from the ones to pass to the job function.

        """
        job = Job(func=func, model_name=model_name, args=args, kwargs=kwargs,
                  priority=priority, eta=eta, max_retries=max_retries, description=description)
        job.user_id = self.session.uid
        self.store(job)
        return job.uuid

    def enqueue_resolve_args(self, func, *args, **kwargs):
        """Create a Job and enqueue it in the queue. Return the job uuid."""
        priority = kwargs.pop('priority', None)
        eta = kwargs.pop('eta', None)
        model_name = kwargs.pop('model_name', None)
        max_retries = kwargs.pop('max_retries', None)
        description = kwargs.pop('description', None)

        return self.enqueue(func, model_name=model_name,
                            args=args, kwargs=kwargs,
                            priority=priority,
                            max_retries=max_retries,
                            eta=eta,
                            description=description)

    def exists(self, job_uuid):
        """Returns if a job still exists in the storage."""
        return bool(self._openerp_id(job_uuid))

    def _openerp_id(self, job_uuid):
        openerp_id = None
        job_ids = self.job_model.search(self.session.cr,
                                        SUPERUSER_ID,
                                        [('uuid', '=', job_uuid)],
                                        context=self.session.context,
                                        limit=1)
        if job_ids:
            openerp_id = job_ids[0]
        return openerp_id

    def openerp_id(self, job):
        return self._openerp_id(job.uuid)

    def _worker_id(self, worker_uuid):
        openerp_id = None
        worker_ids = self.worker_model.search(self.session.cr,
                                              SUPERUSER_ID,
                                              [('uuid', '=', worker_uuid)],
                                              context=self.session.context,
                                              limit=1)
        if worker_ids:
            openerp_id = worker_ids[0]
        return openerp_id

    def store(self, job):
        """ Store the Job """
        vals = {'state': job.state,
                'priority': job.priority,
                'retry': job.retry,
                'max_retries': job.max_retries,
                'exc_info': job.exc_info,
                'user_id': job.user_id or self.session.uid,
                'result': unicode(job.result) if job.result else False,
                'date_enqueued': False,
                'date_started': False,
                'date_done': False,
                'eta': False,
                }

        if job.date_enqueued:
            vals['date_enqueued'] = job.date_enqueued.strftime(
                DEFAULT_SERVER_DATETIME_FORMAT)
        if job.date_started:
            vals['date_started'] = job.date_started.strftime(
                DEFAULT_SERVER_DATETIME_FORMAT)
        if job.date_done:
            vals['date_done'] = job.date_done.strftime(
                DEFAULT_SERVER_DATETIME_FORMAT)
        if job.eta:
            vals['eta'] = job.eta.strftime(DEFAULT_SERVER_DATETIME_FORMAT)

        if job.canceled:
            vals['active'] = False

        if job.worker_uuid:
            vals['worker_id'] = self._worker_id(job.worker_uuid)
        else:
            vals['worker_id'] = False

        if self.exists(job.uuid):
            self.job_model.write(self.session.cr,
                                 self.session.uid,
                                 self.openerp_id(job),
                                 vals,
                                 self.session.context)
        else:
            date_created = job.date_created.strftime(DEFAULT_SERVER_DATETIME_FORMAT)
            vals.update({'uuid': job.uuid,
                         'name': job.description,
                         'func_string': job.func_string,
                         'date_created': date_created,
                         'model_name': (job.model_name if job.model_name
                                        else False),
                         })

            vals['func'] = dumps((job.func_name,
                                  job.args,
                                  job.kwargs))

            self.job_model.create(self.session.cr,
                                  SUPERUSER_ID,
                                  vals,
{'fraction_non_alphanumeric': 0.1066183248325447, 'fraction_numerical': 0.002335156394026916, 'mean_word_length': 1.8510861948142958, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 44, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7423407', 'n_tokens_mistral': 4488, 'n_tokens_neox': 4293, 'n_words': 1175}