source: main/waeup.kofa/trunk/src/waeup/kofa/utils/batching.py @ 14657

Last change on this file since 14657 was 14552, checked in by Henrik Bettermann, 8 years ago

Catch traceback.

  • Property svn:keywords set to Id
File size: 32.7 KB
Line 
1## $Id: batching.py 14552 2017-02-15 06:19:52Z henrik $
2##
3## Copyright (C) 2011 Uli Fouquet & Henrik Bettermann
4## This program is free software; you can redistribute it and/or modify
5## it under the terms of the GNU General Public License as published by
6## the Free Software Foundation; either version 2 of the License, or
7## (at your option) any later version.
8##
9## This program is distributed in the hope that it will be useful,
10## but WITHOUT ANY WARRANTY; without even the implied warranty of
11## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12## GNU General Public License for more details.
13##
14## You should have received a copy of the GNU General Public License
15## along with this program; if not, write to the Free Software
16## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17##
18"""Kofa components for batch processing.
19
20Batch processors eat CSV files to add, update or remove large numbers
21of certain kinds of objects at once.
22"""
23import grok
24import datetime
25import os
26import shutil
27import tempfile
28import time
29import unicodecsv
30import zc.async.interfaces
31from cStringIO import StringIO
32from persistent.list import PersistentList
33from zope.component import createObject, getUtility
34from zope.component.hooks import setSite
35from zope.interface import Interface, implementer
36from zope.schema import getFields
37from zope.schema.interfaces import ConstraintNotSatisfied, RequiredMissing
38from zope.event import notify
39from waeup.kofa.async import AsyncJob
40from waeup.kofa.interfaces import (
41    IBatchProcessor, FatalCSVError, IObjectConverter, IJobManager,
42    ICSVExporter, IGNORE_MARKER, DuplicationError, JOB_STATUS_MAP,
43    IExportJobContainer, IExportJob, IExportContainerFinder)
44
45class BatchProcessor(grok.GlobalUtility):
46    """A processor to add, update, or remove data.
47
48    This is a non-active baseclass.
49    """
50    grok.implements(IBatchProcessor)
51    grok.context(Interface)
52    grok.baseclass()
53
54    # Name used in pages and forms...
55    name = u'Non-registered base processor'
56
57    # Internal name...
58    util_name = ''
59
60    # Items for this processor need an interface with zope.schema fields.
61    iface = Interface
62
63    # The name must be the same as the util_name attribute in order to
64    # register this utility correctly.
65    grok.name(util_name)
66
67    # Headers needed to locate items...
68    location_fields = []
69
70    # A factory with this name must be registered...
71    factory_name = ''
72
73    @property
74    def required_fields(self):
75        """Required fields that have no default.
76
77        A list of names of field, whose value cannot be set if not
78        given during creation. Therefore these fields must exist in
79        input.
80
81        Fields with a default != missing_value do not belong to this
82        category.
83        """
84        result = []
85        for key, field in getFields(self.iface).items():
86            if key in self.location_fields:
87                continue
88            if field.default is not field.missing_value:
89                continue
90            if field.required:
91                result.append(key)
92        return result
93
94    @property
95    def req(self):
96        result = dict(
97            create = self.location_fields + self.required_fields,
98            update = self.location_fields,
99            remove = self.location_fields,
100        )
101        return result
102
103    @property
104    def available_fields(self):
105        return sorted(list(set(
106                    self.location_fields + getFields(self.iface).keys())))
107
108    def getHeaders(self, mode='create'):
109        return self.available_fields
110
111    def checkHeaders(self, headerfields, mode='create'):
112        req = self.req[mode]
113        # Check for required fields...
114        for field in req:
115            if not field in headerfields:
116                raise FatalCSVError(
117                    "Need at least columns %s for import!" %
118                    ', '.join(["'%s'" % x for x in req]))
119        # Check for double fields. Cannot happen because this error is
120        # already catched in views
121        not_ignored_fields = [x for x in headerfields
122                              if not x.startswith('--')]
123        if len(set(not_ignored_fields)) < len(not_ignored_fields):
124            raise FatalCSVError(
125                "Double headers: each column name may only appear once.")
126        return True
127
128    def applyMapping(self, row, mapping):
129        """Apply mapping to a row of CSV data.
130        """
131        result = dict()
132        for key, replacement in mapping.items():
133            if replacement == u'--IGNORE--':
134                # Skip ignored columns in failed and finished data files.
135                continue
136            result[replacement] = row[key]
137        return result
138
139    def getMapping(self, path, headerfields, mode):
140        """Get a mapping from CSV file headerfields to actually used fieldnames.
141
142        """
143        result = dict()
144        reader = unicodecsv.reader(open(path, 'rb'))
145        raw_header = reader.next()
146        for num, field in enumerate(headerfields):
147            if field not in self.location_fields and mode == 'remove':
148                # Skip non-location fields when removing.
149                continue
150            if field == u'--IGNORE--':
151                # Skip ignored columns in failed and finished data files.
152                continue
153            result[raw_header[num]] = field
154        return result
155
156    def stringFromErrs(self, errors, inv_errors):
157        result = []
158        for err in errors:
159            fieldname, message = err
160            result.append("%s: %s" % (fieldname, message))
161        for err in inv_errors:
162            result.append("invariant: %s" % err)
163        return '; '.join(result)
164
165    def callFactory(self, *args, **kw):
166        return createObject(self.factory_name)
167
168    def parentsExist(self, row, site):
169        """Tell whether the parent object for data in ``row`` exists.
170        """
171        raise NotImplementedError('method not implemented')
172
173    def entryExists(self, row, site):
174        """Tell whether there already exists an entry for ``row`` data.
175        """
176        raise NotImplementedError('method not implemented')
177
178    def getParent(self, row, site):
179        """Get the parent object for the entry in ``row``.
180        """
181        raise NotImplementedError('method not implemented')
182
183    def getEntry(self, row, site):
184        """Get the object for the entry in ``row``.
185        """
186        raise NotImplementedError('method not implemented')
187
188    def addEntry(self, obj, row, site):
189        """Add the entry given given by ``row`` data.
190        """
191        raise NotImplementedError('method not implemented')
192
193    def delEntry(self, row, site):
194        """Delete entry given by ``row`` data.
195        """
196        raise NotImplementedError('method not implemented')
197
198    def checkUpdateRequirements(self, obj, row, site):
199        """Checks requirements the object must fulfill when being updated.
200
201        This method is not used in case of deleting or adding objects.
202
203        Returns error messages as strings in case of requirement
204        problems.
205        """
206        return None
207
208    def updateEntry(self, obj, row, site, filename):
209        """Update obj to the values given in row.
210
211        Returns a string describing the fields changed.
212        """
213        changed = []
214        for key, value in row.items():
215            # Skip fields to be ignored.
216            if value == IGNORE_MARKER:
217                continue
218            # Skip fields not declared in interface and which are
219            # not yet attributes of existing objects. We can thus not
220            # add non-existing attributes here.
221            if not hasattr(obj, key):
222                continue
223            # DefaultObjectConverter.fromStringDict fails for
224            # list-of-choices fields because we are using a different
225            # widget for this combination. Thus the ListFieldConverter
226            # returns a useless dictionary which causes getWidgetsData to
227            # skip the field. The value in row remains unchanged.
228            # We have to evaluate the string and replace the value here.
229            try:
230                evalvalue = eval(value)
231                if isinstance(evalvalue, list):
232                    value = evalvalue
233            except:
234                pass
235            try:
236                setattr(obj, key, value)
237            except AttributeError:
238                # Computed attributes can't be set.
239                continue
240            log_value = getattr(value, 'code', value)
241            changed.append('%s=%s' % (key, log_value))
242
243        # If any catalog is involved it must be updated.
244        #
245        # XXX: The event is also triggered when creating objects as
246        # updateEntry is called also when creating entries resulting
247        # in objectAdded and additional objectModified events.
248        if len(changed):
249            notify(grok.ObjectModifiedEvent(obj))
250
251        return ', '.join(changed)
252
253    def createLogfile(self, path, fail_path, num, warnings, mode, user,
254                      timedelta, logger=None):
255        """Write to log file.
256        """
257        if logger is None:
258            return
259        logger.info(
260            "processed: %s, %s mode, %s lines (%s successful/ %s failed), "
261            "%0.3f s (%0.4f s/item)" % (
262            path, mode, num, num - warnings, warnings,
263            timedelta, timedelta/(num or 1)))
264        return
265
266    def writeFailedRow(self, writer, row, warnings):
267        """Write a row with error messages to error CSV.
268
269        If warnings is a list of strings, they will be concatenated.
270        """
271        error_col = warnings
272        if isinstance(warnings, list):
273            error_col = ' / '.join(warnings)
274        row['--ERRORS--'] = error_col
275        writer.writerow(row)
276        return
277
278    def checkConversion(self, row, mode='ignore', ignore_empty=True):
279        """Validates all values in row.
280        """
281        converter = IObjectConverter(self.iface)
282        errs, inv_errs, conv_dict =  converter.fromStringDict(
283            row, self.factory_name, mode=mode)
284        return errs, inv_errs, conv_dict
285
286
287    def emptyRow(self, row):
288        """Detect empty rows.
289        """
290        for value in row.values():
291            if not value in (None, IGNORE_MARKER) and value.strip():
292                return False
293        return True
294
295    def doImport(self, path, headerfields, mode='create', user='Unknown',
296                 logger=None, ignore_empty=True):
297        """In contrast to most other methods, `doImport` is not supposed to
298        be customized, neither in custom packages nor in derived batch
299        processor classes. Therefore, this is the only place where we
300        do import data.
301
302        Before this method starts creating or updating persistent data, it
303        prepares two more files in a temporary folder of the filesystem: (1)
304        a file for pending data with file extension ``.pending`` and (2)
305        a file for successfully processed data with file extension
306        ``.finished``. Then the method starts iterating over all rows of
307        the CSV file. Each row is treated as follows:
308
309        1. An empty row is skipped.
310
311        2. Empty strings or lists (``[]``) in the row are replaced by
312           ignore markers.
313
314        3. The `BatchProcessor.checkConversion` method validates and converts
315           all values in the row. Conversion means the transformation of strings
316           into Python objects. For instance, number expressions have to be
317           transformed into integers, dates into datetime objects, phone number
318           expressions into phone number objects, etc. The converter returns a
319           dictionary with converted values or, if the validation of one of the
320           elements fails, an appropriate warning message. If the conversion
321           fails a pending record is created and stored in the pending data file
322           together with a warning message the converter has raised.
323
324        4. In **create mode** only:
325
326           The parent object must be found and a child
327           object with same object id must not exist. Otherwise the row
328           is skipped, a corresponding warning message is raised and a
329           record is stored in the pending data file.
330
331           Now `doImport` tries to add the new object with the data
332           from the conversion dictionary. In some cases this
333           may fail and a `DuplicationError` is raised. For example, a new
334           payment ticket is created but the same payment for same session
335           has already been made. In this case the object id is unique, no
336           other object with same id exists, but making the 'same' payment
337           twice does not make sense. The import is skipped and a
338           record is stored in the pending data file.
339
340        5. In **update mode** only:
341
342           If the object can't be found, the row is skipped,
343           a ``no such entry`` warning message is raised and a record is
344           stored in the pending data file.
345
346           The `BatchProcessor.checkUpdateRequirements` method checks additional
347           requirements the object must fulfill before being updated. These
348           requirements are not imposed by the data type but the context
349           of the object. For example, post-graduate students have a different
350           registration workflow. With this method we do forbid certain workflow
351           transitions or states.
352
353           Finally, `doImport` updates the existing object with the data
354           from the conversion dictionary.
355
356        6. In **remove mode** only:
357
358           If the object can't be found, the row is skipped,
359           a ``no such entry`` warning message is raised and a record is
360           stored in the pending data file.
361
362           Finally, `doImport` removes the existing object.
363
364        """
365        time_start = time.time()
366        self.checkHeaders(headerfields, mode)
367        mapping = self.getMapping(path, headerfields, mode)
368        reader = unicodecsv.DictReader(open(path, 'rb'))
369
370        temp_dir = tempfile.mkdtemp()
371
372        base = os.path.basename(path)
373        (base, ext) = os.path.splitext(base)
374        failed_path = os.path.join(temp_dir, "%s.pending%s" % (base, ext))
375        failed_headers = mapping.values()
376        failed_headers.append('--ERRORS--')
377        failed_writer = unicodecsv.DictWriter(open(failed_path, 'wb'),
378                                              failed_headers)
379        os.chmod(failed_path, 0664)
380        failed_writer.writerow(dict([(x,x) for x in failed_headers]))
381
382        finished_path = os.path.join(temp_dir, "%s.finished%s" % (base, ext))
383        finished_headers = mapping.values()
384        finished_writer = unicodecsv.DictWriter(open(finished_path, 'wb'),
385                                                finished_headers)
386        os.chmod(finished_path, 0664)
387        finished_writer.writerow(dict([(x,x) for x in finished_headers]))
388
389        num =0
390        num_warns = 0
391        site = grok.getSite()
392
393        for raw_row in reader:
394            num += 1
395            # Skip row if empty
396            if self.emptyRow(raw_row):
397                continue
398            string_row = self.applyMapping(raw_row, mapping)
399            if ignore_empty:
400                # Replace empty strings and empty lists with ignore-markers
401                for key, val in string_row.items():
402                    if val == '' or val == '[]':
403                        string_row[key] = IGNORE_MARKER
404            row = dict(string_row.items()) # create deep copy
405            errs, inv_errs, conv_dict = self.checkConversion(string_row, mode)
406            if errs or inv_errs:
407                num_warns += 1
408                conv_warnings = self.stringFromErrs(errs, inv_errs)
409                self.writeFailedRow(
410                    failed_writer, string_row, conv_warnings)
411                continue
412            row.update(conv_dict)
413
414            if mode == 'create':
415                if not self.parentsExist(row, site):
416                    num_warns += 1
417                    self.writeFailedRow(
418                        failed_writer, string_row,
419                        "Not all parents do exist yet.")
420                    continue
421                if self.entryExists(row, site):
422                    num_warns += 1
423                    self.writeFailedRow(
424                        failed_writer, string_row,
425                        "This object already exists.")
426                    continue
427                obj = self.callFactory()
428                # Override all values in row, also
429                # student_ids and applicant_ids which have been
430                # generated in the respective __init__ methods before.
431                self.updateEntry(obj, row, site, base)
432                try:
433                    self.addEntry(obj, row, site)
434                except KeyError, error:
435                    num_warns += 1
436                    self.writeFailedRow(
437                        failed_writer, string_row, error.message)
438                    continue
439                except DuplicationError, error:
440                    num_warns += 1
441                    self.writeFailedRow(
442                        failed_writer, string_row, error.msg)
443                    continue
444            elif mode == 'remove':
445                if not self.entryExists(row, site):
446                    num_warns += 1
447                    self.writeFailedRow(
448                        failed_writer, string_row,
449                        "Cannot remove: no such entry")
450                    continue
451                self.delEntry(row, site)
452            elif mode == 'update':
453                obj = self.getEntry(row, site)
454                if obj is None:
455                    num_warns += 1
456                    self.writeFailedRow(
457                        failed_writer, string_row,
458                        "Cannot update: no such entry")
459                    continue
460                update_errors = self.checkUpdateRequirements(obj, row, site)
461                if update_errors is not None:
462                    num_warns += 1
463                    self.writeFailedRow(
464                        failed_writer, string_row, update_errors)
465                    continue
466                try:
467                    self.updateEntry(obj, row, site, base)
468                except ConstraintNotSatisfied, err:
469                    num_warns += 1
470                    self.writeFailedRow(
471                        failed_writer, string_row,
472                        "ConstraintNotSatisfied: %s" % err)
473                except RequiredMissing, err:
474                    num_warns += 1
475                    self.writeFailedRow(
476                        failed_writer, string_row,
477                        "RequiredMissing: %s" % err)
478                    continue
479            finished_writer.writerow(string_row)
480
481        time_end = time.time()
482        timedelta = time_end - time_start
483
484        self.createLogfile(path, failed_path, num, num_warns, mode, user,
485                           timedelta, logger=logger)
486        failed_path = os.path.abspath(failed_path)
487        if num_warns == 0:
488            del failed_writer
489            os.unlink(failed_path)
490            failed_path = None
491        return (num, num_warns,
492                os.path.abspath(finished_path), failed_path)
493
494    def get_csv_skeleton(self):
495        """Export CSV file only with a header of available fields.
496
497        A raw string with CSV data should be returned.
498        """
499        outfile = StringIO()
500        writer = unicodecsv.DictWriter(outfile, self.available_fields)
501        writer.writerow(
502            dict(zip(self.available_fields, self.available_fields))) # header
503        outfile.seek(0)
504        return outfile.read()
505
506class ExporterBase(object):
507    """A base for exporters.
508    """
509    grok.implements(ICSVExporter)
510
511    #: Fieldnames considered by this exporter
512    fields = ('code', 'title', 'title_prefix')
513
514    #: The title under which this exporter will be displayed
515    #: (if registered as a utility)
516    title = 'Override this title'
517
518    def mangle_value(self, value, name, context=None):
519        """Hook for mangling values in derived classes.
520        """
521        if isinstance(value, bool):
522            value = value and '1' or '0'
523        elif isinstance(value, unicode):
524            # CSV writers like byte streams better than unicode
525            value = value.encode('utf-8')
526        elif isinstance(value, datetime.datetime):
527            #value = str(value)
528            value = str('%s#' % value) # changed 2014-07-06, see ticket #941
529        elif isinstance(value, datetime.date):
530            # Order is important here: check for date after datetime as
531            # datetimes are also dates.
532            #
533            # Append hash '#' to dates to circumvent unwanted excel automatic
534            value = str('%s#' % value)
535        elif value is None:
536            # None is not really representable in CSV files
537            value = ''
538        return value
539
540    def get_csv_writer(self, filepath=None):
541        """Get a CSV dict writer instance open for writing.
542
543        Returns a tuple (<writer>, <outfile>) where ``<writer>`` is a
544        :class:`csv.DictWriter` instance and outfile is the real file
545        which is written to. The latter is important when writing to
546        StringIO and can normally be ignored otherwise.
547
548        The returned file will already be filled with the header row.
549
550        Please note that if you give a filepath, the returned outfile
551        is open for writing only and you might have to close it before
552        reopening it for reading.
553        """
554        if filepath is None:
555            outfile = StringIO()
556        else:
557            outfile = open(filepath, 'wb')
558        writer = unicodecsv.DictWriter(outfile, self.fields)
559        writer.writerow(dict(zip(self.fields, self.fields))) # header
560        return writer, outfile
561
562    def write_item(self, obj, writer):
563        """Write a row extracted from `obj` into CSV file using `writer`.
564        """
565        row = {}
566        for name in self.fields:
567            value = getattr(obj, name, None)
568            value = self.mangle_value(value, name, obj)
569            row[name] = value
570        writer.writerow(row)
571        return
572
573    def close_outfile(self, filepath, outfile):
574        """Close outfile.
575        If filepath is None, the contents of outfile is returned.
576        """
577        outfile.seek(0)
578        if filepath is None:
579            return outfile.read()
580        outfile.close()
581        return
582
583    def get_filtered(self, site, **kw):
584        """Get datasets to export filtered by keyword arguments.
585        Returns an iterable.
586        """
587        raise NotImplementedError
588
589    def get_selected(self, site, selected):
590        """Get datasets to export for selected items
591        specified by a list of identifiers.
592        Returns an iterable.
593        """
594        raise NotImplementedError
595
596    def export(self, iterable, filepath=None):
597        """Export `iterable` as CSV file.
598        If `filepath` is ``None``, a raw string with CSV data should
599        be returned.
600        """
601        raise NotImplementedError
602
603    def export_all(self, site, filepath=None):
604        """Export all appropriate objects in `site` into `filepath` as
605        CSV data.
606        If `filepath` is ``None``, a raw string with CSV data should
607        be returned.
608        """
609        raise NotImplementedError
610
611    def export_filtered(self, site, filepath=None, **kw):
612        """Export items denoted by `kw`.
613        If `filepath` is ``None``, a raw string with CSV data should
614        be returned.
615        """
616        data = self.get_filtered(site, **kw)
617        return self.export(data, filepath=filepath)
618
619    def export_selected(self, site, filepath=None, **kw):
620        """Export those items specified by a list of identifiers
621        called `selected`.
622        If `filepath` is ``None``, a raw string with CSV data should
623        be returned.
624        """
625        selected = kw.get('selected', [])
626        data = self.get_selected(site, selected)
627        return self.export(data, filepath=filepath)
628
629def export_job(site, exporter_name, **kw):
630    """Export all entries delivered by exporter and store it in a temp file.
631
632    `site` gives the site to search. It will be passed to the exporter
633    and also be set as 'current site' as the function is used in
634    asynchronous jobs which run in their own threads and have no site
635    set initially. Therefore `site` must also be a valid value for use
636    with `zope.component.hooks.setSite()`.
637
638    `exporter_name` is the utility name under which the desired
639    exporter was registered with the ZCA.
640
641    The resulting CSV file will be stored in a new temporary directory
642    (using :func:`tempfile.mkdtemp`). It will be named after the
643    exporter used with `.csv` filename extension.
644
645    Returns the path to the created CSV file.
646
647    .. note:: It is the callers responsibility to clean up the used
648              file and its parent directory.
649    """
650    setSite(site)
651    exporter = getUtility(ICSVExporter, name=exporter_name)
652    output_dir = tempfile.mkdtemp()
653    filename = '%s.csv' % exporter_name
654    output_path = os.path.join(output_dir, filename)
655    if kw == {}:
656        exporter.export_all(site, filepath=output_path)
657    elif kw.has_key('selected'):
658        exporter.export_selected(site, filepath=output_path, **kw)
659    else:
660        exporter.export_filtered(site, filepath=output_path, **kw)
661    return output_path
662
663class AsyncExportJob(AsyncJob):
664    """An IJob that exports data to CSV files.
665
666    `AsyncExportJob` instances are regular `AsyncJob` instances with a
667    different constructor API. Instead of a callable to execute, you
668    must pass a `site` and some `exporter_name` to trigger an export.
669
670    The real work is done when an instance of this class is put into a
671    queue. See :mod:`waeup.kofa.async` to learn more about
672    asynchronous jobs.
673
674    The `exporter_name` must be the name under which an ICSVExporter
675    utility was registered with the ZCA.
676
677    The `site` must be a valid site  or ``None``.
678
679    The result of an `AsyncExportJob` is the path to generated CSV
680    file. The file will reside in a temporary directory that should be
681    removed after being used.
682    """
683    grok.implements(IExportJob)
684
685    def __init__(self, site, exporter_name, *args, **kwargs):
686        super(AsyncExportJob, self).__init__(
687            export_job, site, exporter_name, *args, **kwargs)
688
689    @property
690    def finished(self):
691        """A job is marked `finished` if it is completed.
692
693        Please note: a finished report job does not neccessarily
694        provide an IReport result. See meth:`failed`.
695        """
696        return self.status == zc.async.interfaces.COMPLETED
697
698    @property
699    def failed(self):
700        """A report job is marked failed iff it is finished and the
701        result is None.
702
703        While a job is unfinished, the `failed` status is ``None``.
704
705        Failed jobs normally provide a `traceback` to examine reasons.
706        """
707        if not self.finished:
708            return None
709        if getattr(self, 'result', None) is None:
710            return True
711        return False
712
713class ExportJobContainer(object):
714    """A mix-in that provides functionality for asynchronous export jobs.
715    """
716    grok.implements(IExportJobContainer)
717    running_exports = PersistentList()
718
719    def start_export_job(self, exporter_name, user_id, *args, **kwargs):
720        """Start asynchronous export job.
721
722        `exporter_name` is the name of an exporter utility to be used.
723
724        `user_id` is the ID of the user that triggers the export.
725
726        The job_id is stored along with exporter name and user id in a
727        persistent list.
728
729        The method supports additional positional and keyword
730        arguments, which are passed as-is to the respective
731        :class:`AsyncExportJob`.
732
733        Returns the job ID of the job started.
734        """
735        site = grok.getSite()
736        manager = getUtility(IJobManager)
737        job = AsyncExportJob(site, exporter_name, *args, **kwargs)
738        job_id = manager.put(job)
739        # Make sure that the persisted list is stored in ZODB
740        self.running_exports = PersistentList(self.running_exports)
741        self.running_exports.append((job_id, exporter_name, user_id))
742        return job_id
743
744    def get_running_export_jobs(self, user_id=None):
745        """Get export jobs for user with `user_id` as list of tuples.
746
747        Each tuples holds ``<job_id>, <exporter_name>, <user_id>`` in
748        that order. The ``<exporter_name>`` is the utility name of the
749        used exporter.
750
751        If `user_id` is ``None``, all running jobs are returned.
752        """
753        entries = []
754        to_delete = []
755        manager = getUtility(IJobManager)
756        for entry in self.running_exports:
757            if user_id is not None and entry[2] != user_id:
758                continue
759            if manager.get(entry[0]) is None:
760                to_delete.append(entry)
761                continue
762            entries.append(entry)
763        if to_delete:
764            self.running_exports = PersistentList(
765                [x for x in self.running_exports if x not in to_delete])
766        return entries
767
768    def get_export_jobs_status(self, user_id=None):
769        """Get running/completed export jobs for `user_id` as list of tuples.
770
771        Each tuple holds ``<raw status>, <status translated>,
772        <exporter title>`` in that order, where ``<status
773        translated>`` and ``<exporter title>`` are translated strings
774        representing the status of the job and the human readable
775        title of the exporter used.
776        """
777        entries = self.get_running_export_jobs(user_id)
778        result = []
779        manager = getUtility(IJobManager)
780        for entry in entries:
781            job = manager.get(entry[0])
782            if job is None:
783                continue
784            status, status_translated = JOB_STATUS_MAP[job.status]
785            exporter_name = getUtility(ICSVExporter, name=entry[1]).title
786            result.append((status, status_translated, exporter_name))
787        return result
788
789    def delete_export_entry(self, entry):
790        """Delete the export denoted by `entry`.
791
792        Removes given entry from the local `running_exports` list and also
793        removes the regarding job via the local job manager.
794
795        `entry` must be a tuple ``(<job id>, <exporter name>, <user
796        id>)`` as created by :meth:`start_export_job` or returned by
797        :meth:`get_running_export_jobs`.
798        """
799        manager = getUtility(IJobManager)
800        job = manager.get(entry[0])
801        if job is not None:
802            # remove created export file
803            if isinstance(job.result, basestring):
804                if os.path.exists(os.path.dirname(job.result)):
805                    shutil.rmtree(os.path.dirname(job.result))
806        manager.remove(entry[0], self)
807        new_entries = [x for x in self.running_exports
808                       if x != entry]
809        self.running_exports = PersistentList(new_entries)
810        return
811
812    def entry_from_job_id(self, job_id):
813        """Get entry tuple for `job_id`.
814
815        Returns ``None`` if no such entry can be found.
816        """
817        for entry in self.running_exports:
818            if entry[0] == job_id:
819                return entry
820        return None
821
822class VirtualExportJobContainer(ExportJobContainer):
823    """A virtual export job container.
824
825    Virtual ExportJobContainers can be used as a mixin just like real
826    ExportJobContainer.
827
828    They retrieve and store data in the site-wide ExportJobContainer.
829
830    Functionality is currently entirely as for regular
831    ExportJobContainers, except that data is stored elsewhere.
832
833    VirtualExportJobContainers need a registered
834    IExportContainerFinder utility to find a suitable container for
835    storing data.
836    """
837    grok.implements(IExportJobContainer)
838
839    @property
840    def _site_container(self):
841        return getUtility(IExportContainerFinder)()
842
843    # The following is a simple trick. While ExportJobContainers store
844    # only one attribute in ZODB, it is sufficient to replace this
845    # attribute `running_exports` with a suitable manager to make the
846    # whole virtual container work like the original but with the data
847    # stored in the site-wide exports container. This way, virtual
848    # export containers provide the whole functionality of a regular
849    # exports container but store no data at all with themselves.
850    @property
851    def running_exports(self):
852        """Exports stored in the site-wide exports container.
853        """
854        return self._site_container.running_exports
855
856    @running_exports.setter
857    def running_exports(self, value):
858        self._site_container.running_exports = value
859
860    @running_exports.deleter
861    def running_exports(self):
862        del self._site_container.running_exports
863
864    @property
865    def logger(self):
866        return self._site_container.logger
867
868@implementer(IExportContainerFinder)
869class ExportContainerFinder(grok.GlobalUtility):
870    """Finder for local (site-wide) export container.
871    """
872
873    def __call__(self):
874        """Get the local export container-
875
876        If no site can be determined or the site provides no export
877        container, None is returned.
878        """
879        site = grok.getSite()
880        if site is None:
881            return None
882        return site.get('datacenter', None)
Note: See TracBrowser for help on using the repository browser.