Changeset 9089 for main/waeup.kofa/branches/uli-zc-async/src/waeup

Timestamp:

8 Aug 2012, 13:19:41 (12 years ago)

Author:

uli

Message:

Add async functionality for exporters:

• an exporter function usable with decoupled threads
• a mix-in class for components storing export jobs
• specialized async-aware job implementation

File:

• main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/utils/batching.py (modified) (2 diffs)

main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/utils/batching.py

@@ -25 +25 @@
 import datetime
 import os
+import shutil
 import tempfile
 import time
 from cStringIO import StringIO
-from zope.component import createObject
+from persistent.list import PersistentList
+from zope.component import createObject, getUtility
+from zope.component.hooks import setSite
 from zope.interface import Interface
 from zope.schema import getFields
 from zope.event import notify
+from waeup.kofa.async import AsyncJob
 from waeup.kofa.interfaces import (
-    IBatchProcessor, FatalCSVError, IObjectConverter,
-    ICSVExporter, IGNORE_MARKER, DuplicationError)
+    IBatchProcessor, FatalCSVError, IObjectConverter, IJobManager,
+    ICSVExporter, IGNORE_MARKER, DuplicationError, JOB_STATUS_MAP)
 
 class BatchProcessor(grok.GlobalUtility):
@@ -486 +490 @@
         """
         raise NotImplementedError
+
+
+def export_job(site, exporter_name):
+    """Export all entries delivered by exporter and store it in a temp file.
+
+    `site` gives the site to search. It will be passed to the exporter
+    and also be set as 'current site' as the function is used in
+    asynchronous jobs which run in their own threads and have no site
+    set initially. Therefore `site` must also be a valid value for use
+    with `zope.component.hooks.setSite()`.
+
+    `exporter_name` is the utility name under which the desired
+    exporter was registered with the ZCA.
+
+    The resulting CSV file will be stored in a new temporary directory
+    (using :func:`tempfile.mkdtemp`).
+
+    Returns the path to the created CSV file.
+
+    .. note:: It is the callers responsibility to clean up the used
+              file and its parent directory.
+    """
+    setSite(site)
+    exporter = getUtility(ICSVExporter, name=exporter_name)
+    output_dir = tempfile.mkdtemp()
+    output_path = os.path.join(output_dir, 'export.csv')
+    exporter.export_all(site, filepath=output_path)
+    return output_path
+
+class AsyncExportJob(AsyncJob):
+
+    def __init__(self, site, exporter_name):
+        super(AsyncExportJob, self).__init__(
+            export_job, site, exporter_name)
+
+class ExportJobContainer(object):
+    """A mix-in that provides functionality for asynchronous export jobs.
+    """
+    running_exports = PersistentList()
+
+    def start_export_job(self, exporter_name, user_id):
+        """Start asynchronous export job.
+
+        `exporter_name` is the name of an exporter utility to be used.
+
+        `user_id` is the ID of the user that triggers the export.
+
+        The job_id is stored along with exporter name and user id in a
+        persistent list.
+
+        Returns the job ID of the job started.
+        """
+        site = grok.getSite()
+        manager = getUtility(IJobManager)
+        job = AsyncExportJob(site, exporter_name)
+        job_id = manager.put(job)
+        self.running_exports.append((job_id, exporter_name, user_id))
+        # Make sure that the persisted list is stored in ZODB
+        self.running_exports = PersistentList(self.running_exports)
+        return job_id
+
+    def get_running_export_jobs(self, user_id=None):
+        """Get export jobs for user with `user_id` as list of tuples.
+
+        Each tuples holds ``<job_id>, <exporter_name>, <user_id>`` in
+        that order. The ``<exporter_name>`` is the utility name of the
+        used exporter.
+
+        If `user_id` is ``None``, all running jobs are returned.
+        """
+        entries = []
+        to_delete = []
+        manager = getUtility(IJobManager)
+        for entry in self.running_exports:
+            if user_id is not None and entry[2] != user_id:
+                continue
+            if manager.get(entry[0]) is None:
+                to_delete.append(entry)
+                continue
+            entries.append(entry)
+        if to_delete:
+            self.running_exports = PersistentList(
+                [x for x in self.running_exports if x not in to_delete])
+        return entries
+
+    def get_export_jobs_status(self, user_id=None):
+        """Get running/completed export jobs for `user_id` as list of tuples.
+
+        Each tuple holds ``<raw status>, <status translated>,
+        <exporter title>`` in that order, where ``<status
+        translated>`` and ``<exporter title>`` are translated strings
+        representing the status of the job and the human readable
+        title of the exporter used.
+        """
+        entries = self.get_running_export_jobs(user_id)
+        result = []
+        manager = getUtility(IJobManager)
+        for entry in entries:
+            job = manager.get(entry[0])
+            if job is None:
+                continue
+            status, status_translated = JOB_STATUS_MAP[job.status]
+            exporter_name = getUtility(ICSVExporter, name=entry[1]).title
+            result.append((status, status_translated, exporter_name))
+        return result
+
+    def delete_export_entry(self, entry):
+        """Delete the export denoted by `entry`.
+
+        Removes `entry` from the local `running_exports` list and also
+        removes the regarding job via the local job manager.
+
+        `entry` is a tuple ``(<job id>, <exporter name>, <user id>)``
+        as created by :meth:`start_export_job` or returned by
+        :meth:`get_running_export_jobs`.
+        """
+        manager = getUtility(IJobManager)
+        job = manager.get(entry[0])
+        if job is not None:
+            # remove created export file
+            if isinstance(job.result, basestring):
+                if os.path.exists(os.path.dirname(job.result)):
+                    shutil.rmtree(os.path.dirname(job.result))
+        manager.remove(entry[0], self)
+        new_entries = [x for x in self.running_exports
+                       if x != entry]
+        self.running_exports = PersistentList(new_entries)
+        return