Produits Entr'ouvert » w.c.s.

    python3-uwsgidecorators,

WCS_MANAGE_COMMAND = '/usr/bin/wcs-manage'

spooler-processes = 3

import = wcs.qommon.spooler

var/lib/wcs/spooler

--spooler /var/lib/wcs/spooler/

    chown $USER:$GROUP /var/lib/$NAME/spooler

ExecStartPre=/bin/mkdir -p /var/lib/wcs/spooler/%m/

ExecStart=/usr/bin/uwsgi --ini /etc/%p/uwsgi.ini --spooler /var/lib/wcs/spooler/%m/

from wcs.qommon.afterjobs import AfterJob

class TestAfterJob(AfterJob):

    def execute(self):

        pass

def test_runjob(pub):

    with pytest.raises(CommandError):

        call_command('runjob')

    with pytest.raises(CommandError):

        call_command('runjob', '--domain=example.net', '--job-id=%s' % 'invalid')

    job = TestAfterJob(label='test')

    job.store()

    assert AfterJob.get(job.id).status == 'registered'

    call_command('runjob', '--domain=example.net', '--job-id=%s' % job.id)

    assert AfterJob.get(job.id).status == 'completed'

    job1 = AfterJob()

    job1.status = 'completed'

    job1.creation_time = time.time() - 3 * 3600

    job1.completion_time = time.time() - 3 * 3600

    job1.store()

    job2 = AfterJob()

    job2.status = 'completed'

    job2.creation_time = time.time()

    job2.completion_time = time.time()

    job2.store()

    job3 = AfterJob()

    job3.status = 'running'

    job3.creation_time = time.time() - 3 * 86400

    job3.store()

    assert AfterJob.select()[0].id == job2.id

# w.c.s. - web application for online forms

# Copyright (C) 2005-2020  Entr'ouvert

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 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 General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, see <http://www.gnu.org/licenses/>.

import sys

from django.core.management import CommandError

from wcs.qommon.http_response import AfterJob

from . import TenantCommand

class Command(TenantCommand):

    '''Run an afterjob (internal command)'''

    def add_arguments(self, parser):

        parser.add_argument('--domain', action='store', required=True)

        parser.add_argument('--job-id', action='store', required=True)

    def handle(self, *args, **options):

        domain = options.pop('domain')

        self.init_tenant_publisher(domain)

        try:

            job = AfterJob.get(options['job_id'])

        except KeyError:

            raise CommandError('missing job')

        job.run()

import sys

import time

import traceback

import uuid

from quixote import get_publisher, get_response

from . import _, N_, errors

from .storage import StorableObject

class AfterJob(StorableObject):

    _names = 'afterjobs'

    _reset_class = False

    label = None

    status = None

    creation_time = None

    completion_time = None

    completion_status = None

    execute = None

    def __init__(self, label=None, cmd=None, **kwargs):

        super().__init__(id=str(uuid.uuid4()))

        if label:

            self.label = label

        self.creation_time = time.time()

        self.job_cmd = cmd

        self.status = N_('registered')

        self.kwargs = kwargs

    def run(self, spool=False):

        if self.completion_time:

            return

        if spool and self.id and self.execute:

            from django.conf import settings

            if 'uwsgi' in sys.modules and settings.WCS_MANAGE_COMMAND:

                from .spooler import run_after_job

                self.store()

                run_after_job.spool(tenant_dir=get_publisher().app_dir, job_id=self.id)

                return

        self.status = N_('running')

        if self.id:

            self.store()

        try:

            if self.execute:

                self.execute()

            else:

                self.job_cmd(job=self)

        except Exception:

            get_publisher().notify_of_exception(sys.exc_info())

            self.exception = traceback.format_exc()

            self.status = N_('failed')

        else:

            self.status = N_('completed')

        self.completion_time = time.time()

        if self.id:

            self.store()

    def __getstate__(self):

        if not isinstance(self.job_cmd, str):

            obj_dict = self.__dict__.copy()

            obj_dict['job_cmd'] = None

            return obj_dict

        return self.__dict__

from .afterjobs import AfterJob

    def add_after_job(self, label_or_instance, cmd=None, fire_and_forget=False):

        if isinstance(label_or_instance, AfterJob):

            job = label_or_instance

        else:

            job = AfterJob(label=label_or_instance, cmd=cmd)

        self.after_jobs.append(job)

        for job in self.after_jobs or []:

            job.run(spool=True)

# w.c.s. - web application for online forms

# Copyright (C) 2005-2020  Entr'ouvert

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 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 General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, see <http://www.gnu.org/licenses/>.

import subprocess

from uwsgidecorators import spool

@spool

def run_after_job(args):

    from django.conf import settings

    subprocess.run([

        settings.WCS_MANAGE_COMMAND,

        'runjob',

        '--domain', args['tenant_dir'].strip('/').split('/')[-1],

        '--job-id', args['job_id']

    ])

    _reset_class = True  # reset loaded object class

        if cls._reset_class:

            o.__class__ = cls

# management command, used to run afterjobs in uwsgi mode

WCS_MANAGE_COMMAND = None