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

Last change on this file since 14194 was 13159, checked in by Henrik Bettermann, 9 years ago

Enable import of list-of-choices fields.

  • Property svn:keywords set to Id
File size: 32.5 KB
Line 
1## $Id: batching.py 13159 2015-07-10 11:33:03Z 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
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 value.strip() and not value in (None, IGNORE_MARKER):
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                    continue
474            finished_writer.writerow(string_row)
475
476        time_end = time.time()
477        timedelta = time_end - time_start
478
479        self.createLogfile(path, failed_path, num, num_warns, mode, user,
480                           timedelta, logger=logger)
481        failed_path = os.path.abspath(failed_path)
482        if num_warns == 0:
483            del failed_writer
484            os.unlink(failed_path)
485            failed_path = None
486        return (num, num_warns,
487                os.path.abspath(finished_path), failed_path)
488
489    def get_csv_skeleton(self):
490        """Export CSV file only with a header of available fields.
491
492        A raw string with CSV data should be returned.
493        """
494        outfile = StringIO()
495        writer = unicodecsv.DictWriter(outfile, self.available_fields)
496        writer.writerow(
497            dict(zip(self.available_fields, self.available_fields))) # header
498        outfile.seek(0)
499        return outfile.read()
500
501class ExporterBase(object):
502    """A base for exporters.
503    """
504    grok.implements(ICSVExporter)
505
506    #: Fieldnames considered by this exporter
507    fields = ('code', 'title', 'title_prefix')
508
509    #: The title under which this exporter will be displayed
510    #: (if registered as a utility)
511    title = 'Override this title'
512
513    def mangle_value(self, value, name, context=None):
514        """Hook for mangling values in derived classes.
515        """
516        if isinstance(value, bool):
517            value = value and '1' or '0'
518        elif isinstance(value, unicode):
519            # CSV writers like byte streams better than unicode
520            value = value.encode('utf-8')
521        elif isinstance(value, datetime.datetime):
522            #value = str(value)
523            value = str('%s#' % value) # changed 2014-07-06, see ticket #941
524        elif isinstance(value, datetime.date):
525            # Order is important here: check for date after datetime as
526            # datetimes are also dates.
527            #
528            # Append hash '#' to dates to circumvent unwanted excel automatic
529            value = str('%s#' % value)
530        elif value is None:
531            # None is not really representable in CSV files
532            value = ''
533        return value
534
535    def get_csv_writer(self, filepath=None):
536        """Get a CSV dict writer instance open for writing.
537
538        Returns a tuple (<writer>, <outfile>) where ``<writer>`` is a
539        :class:`csv.DictWriter` instance and outfile is the real file
540        which is written to. The latter is important when writing to
541        StringIO and can normally be ignored otherwise.
542
543        The returned file will already be filled with the header row.
544
545        Please note that if you give a filepath, the returned outfile
546        is open for writing only and you might have to close it before
547        reopening it for reading.
548        """
549        if filepath is None:
550            outfile = StringIO()
551        else:
552            outfile = open(filepath, 'wb')
553        writer = unicodecsv.DictWriter(outfile, self.fields)
554        writer.writerow(dict(zip(self.fields, self.fields))) # header
555        return writer, outfile
556
557    def write_item(self, obj, writer):
558        """Write a row extracted from `obj` into CSV file using `writer`.
559        """
560        row = {}
561        for name in self.fields:
562            value = getattr(obj, name, None)
563            value = self.mangle_value(value, name, obj)
564            row[name] = value
565        writer.writerow(row)
566        return
567
568    def close_outfile(self, filepath, outfile):
569        """Close outfile.
570        If filepath is None, the contents of outfile is returned.
571        """
572        outfile.seek(0)
573        if filepath is None:
574            return outfile.read()
575        outfile.close()
576        return
577
578    def get_filtered(self, site, **kw):
579        """Get datasets to export filtered by keyword arguments.
580        Returns an iterable.
581        """
582        raise NotImplementedError
583
584    def get_selected(self, site, selected):
585        """Get datasets to export for selected items
586        specified by a list of identifiers.
587        Returns an iterable.
588        """
589        raise NotImplementedError
590
591    def export(self, iterable, filepath=None):
592        """Export `iterable` as CSV file.
593        If `filepath` is ``None``, a raw string with CSV data should
594        be returned.
595        """
596        raise NotImplementedError
597
598    def export_all(self, site, filepath=None):
599        """Export all appropriate objects in `site` into `filepath` as
600        CSV data.
601        If `filepath` is ``None``, a raw string with CSV data should
602        be returned.
603        """
604        raise NotImplementedError
605
606    def export_filtered(self, site, filepath=None, **kw):
607        """Export items denoted by `kw`.
608        If `filepath` is ``None``, a raw string with CSV data should
609        be returned.
610        """
611        data = self.get_filtered(site, **kw)
612        return self.export(data, filepath=filepath)
613
614    def export_selected(self, site, filepath=None, **kw):
615        """Export those items specified by a list of identifiers
616        called `selected`.
617        If `filepath` is ``None``, a raw string with CSV data should
618        be returned.
619        """
620        selected = kw.get('selected', [])
621        data = self.get_selected(site, selected)
622        return self.export(data, filepath=filepath)
623
624def export_job(site, exporter_name, **kw):
625    """Export all entries delivered by exporter and store it in a temp file.
626
627    `site` gives the site to search. It will be passed to the exporter
628    and also be set as 'current site' as the function is used in
629    asynchronous jobs which run in their own threads and have no site
630    set initially. Therefore `site` must also be a valid value for use
631    with `zope.component.hooks.setSite()`.
632
633    `exporter_name` is the utility name under which the desired
634    exporter was registered with the ZCA.
635
636    The resulting CSV file will be stored in a new temporary directory
637    (using :func:`tempfile.mkdtemp`). It will be named after the
638    exporter used with `.csv` filename extension.
639
640    Returns the path to the created CSV file.
641
642    .. note:: It is the callers responsibility to clean up the used
643              file and its parent directory.
644    """
645    setSite(site)
646    exporter = getUtility(ICSVExporter, name=exporter_name)
647    output_dir = tempfile.mkdtemp()
648    filename = '%s.csv' % exporter_name
649    output_path = os.path.join(output_dir, filename)
650    if kw == {}:
651        exporter.export_all(site, filepath=output_path)
652    elif kw.has_key('selected'):
653        exporter.export_selected(site, filepath=output_path, **kw)
654    else:
655        exporter.export_filtered(site, filepath=output_path, **kw)
656    return output_path
657
658class AsyncExportJob(AsyncJob):
659    """An IJob that exports data to CSV files.
660
661    `AsyncExportJob` instances are regular `AsyncJob` instances with a
662    different constructor API. Instead of a callable to execute, you
663    must pass a `site` and some `exporter_name` to trigger an export.
664
665    The real work is done when an instance of this class is put into a
666    queue. See :mod:`waeup.kofa.async` to learn more about
667    asynchronous jobs.
668
669    The `exporter_name` must be the name under which an ICSVExporter
670    utility was registered with the ZCA.
671
672    The `site` must be a valid site  or ``None``.
673
674    The result of an `AsyncExportJob` is the path to generated CSV
675    file. The file will reside in a temporary directory that should be
676    removed after being used.
677    """
678    grok.implements(IExportJob)
679
680    def __init__(self, site, exporter_name, *args, **kwargs):
681        super(AsyncExportJob, self).__init__(
682            export_job, site, exporter_name, *args, **kwargs)
683
684    @property
685    def finished(self):
686        """A job is marked `finished` if it is completed.
687
688        Please note: a finished report job does not neccessarily
689        provide an IReport result. See meth:`failed`.
690        """
691        return self.status == zc.async.interfaces.COMPLETED
692
693    @property
694    def failed(self):
695        """A report job is marked failed iff it is finished and the
696        result is None.
697
698        While a job is unfinished, the `failed` status is ``None``.
699
700        Failed jobs normally provide a `traceback` to examine reasons.
701        """
702        if not self.finished:
703            return None
704        if getattr(self, 'result', None) is None:
705            return True
706        return False
707
708class ExportJobContainer(object):
709    """A mix-in that provides functionality for asynchronous export jobs.
710    """
711    grok.implements(IExportJobContainer)
712    running_exports = PersistentList()
713
714    def start_export_job(self, exporter_name, user_id, *args, **kwargs):
715        """Start asynchronous export job.
716
717        `exporter_name` is the name of an exporter utility to be used.
718
719        `user_id` is the ID of the user that triggers the export.
720
721        The job_id is stored along with exporter name and user id in a
722        persistent list.
723
724        The method supports additional positional and keyword
725        arguments, which are passed as-is to the respective
726        :class:`AsyncExportJob`.
727
728        Returns the job ID of the job started.
729        """
730        site = grok.getSite()
731        manager = getUtility(IJobManager)
732        job = AsyncExportJob(site, exporter_name, *args, **kwargs)
733        job_id = manager.put(job)
734        # Make sure that the persisted list is stored in ZODB
735        self.running_exports = PersistentList(self.running_exports)
736        self.running_exports.append((job_id, exporter_name, user_id))
737        return job_id
738
739    def get_running_export_jobs(self, user_id=None):
740        """Get export jobs for user with `user_id` as list of tuples.
741
742        Each tuples holds ``<job_id>, <exporter_name>, <user_id>`` in
743        that order. The ``<exporter_name>`` is the utility name of the
744        used exporter.
745
746        If `user_id` is ``None``, all running jobs are returned.
747        """
748        entries = []
749        to_delete = []
750        manager = getUtility(IJobManager)
751        for entry in self.running_exports:
752            if user_id is not None and entry[2] != user_id:
753                continue
754            if manager.get(entry[0]) is None:
755                to_delete.append(entry)
756                continue
757            entries.append(entry)
758        if to_delete:
759            self.running_exports = PersistentList(
760                [x for x in self.running_exports if x not in to_delete])
761        return entries
762
763    def get_export_jobs_status(self, user_id=None):
764        """Get running/completed export jobs for `user_id` as list of tuples.
765
766        Each tuple holds ``<raw status>, <status translated>,
767        <exporter title>`` in that order, where ``<status
768        translated>`` and ``<exporter title>`` are translated strings
769        representing the status of the job and the human readable
770        title of the exporter used.
771        """
772        entries = self.get_running_export_jobs(user_id)
773        result = []
774        manager = getUtility(IJobManager)
775        for entry in entries:
776            job = manager.get(entry[0])
777            if job is None:
778                continue
779            status, status_translated = JOB_STATUS_MAP[job.status]
780            exporter_name = getUtility(ICSVExporter, name=entry[1]).title
781            result.append((status, status_translated, exporter_name))
782        return result
783
784    def delete_export_entry(self, entry):
785        """Delete the export denoted by `entry`.
786
787        Removes given entry from the local `running_exports` list and also
788        removes the regarding job via the local job manager.
789
790        `entry` must be a tuple ``(<job id>, <exporter name>, <user
791        id>)`` as created by :meth:`start_export_job` or returned by
792        :meth:`get_running_export_jobs`.
793        """
794        manager = getUtility(IJobManager)
795        job = manager.get(entry[0])
796        if job is not None:
797            # remove created export file
798            if isinstance(job.result, basestring):
799                if os.path.exists(os.path.dirname(job.result)):
800                    shutil.rmtree(os.path.dirname(job.result))
801        manager.remove(entry[0], self)
802        new_entries = [x for x in self.running_exports
803                       if x != entry]
804        self.running_exports = PersistentList(new_entries)
805        return
806
807    def entry_from_job_id(self, job_id):
808        """Get entry tuple for `job_id`.
809
810        Returns ``None`` if no such entry can be found.
811        """
812        for entry in self.running_exports:
813            if entry[0] == job_id:
814                return entry
815        return None
816
817class VirtualExportJobContainer(ExportJobContainer):
818    """A virtual export job container.
819
820    Virtual ExportJobContainers can be used as a mixin just like real
821    ExportJobContainer.
822
823    They retrieve and store data in the site-wide ExportJobContainer.
824
825    Functionality is currently entirely as for regular
826    ExportJobContainers, except that data is stored elsewhere.
827
828    VirtualExportJobContainers need a registered
829    IExportContainerFinder utility to find a suitable container for
830    storing data.
831    """
832    grok.implements(IExportJobContainer)
833
834    @property
835    def _site_container(self):
836        return getUtility(IExportContainerFinder)()
837
838    # The following is a simple trick. While ExportJobContainers store
839    # only one attribute in ZODB, it is sufficient to replace this
840    # attribute `running_exports` with a suitable manager to make the
841    # whole virtual container work like the original but with the data
842    # stored in the site-wide exports container. This way, virtual
843    # export containers provide the whole functionality of a regular
844    # exports container but store no data at all with themselves.
845    @property
846    def running_exports(self):
847        """Exports stored in the site-wide exports container.
848        """
849        return self._site_container.running_exports
850
851    @running_exports.setter
852    def running_exports(self, value):
853        self._site_container.running_exports = value
854
855    @running_exports.deleter
856    def running_exports(self):
857        del self._site_container.running_exports
858
859    @property
860    def logger(self):
861        return self._site_container.logger
862
863@implementer(IExportContainerFinder)
864class ExportContainerFinder(grok.GlobalUtility):
865    """Finder for local (site-wide) export container.
866    """
867
868    def __call__(self):
869        """Get the local export container-
870
871        If no site can be determined or the site provides no export
872        container, None is returned.
873        """
874        site = grok.getSite()
875        if site is None:
876            return None
877        return site.get('datacenter', None)
Note: See TracBrowser for help on using the repository browser.