Changeset 9217 for main/waeup.kofa/trunk/src/waeup/kofa/utils

Timestamp:

21 Sep 2012, 11:21:05 (12 years ago)

Author:

uli

Message:

Merge changes from uli-async-update back into trunk.

Location:

main/waeup.kofa/trunk/src/waeup/kofa/utils

Files:

• batching.py (modified) (3 diffs)
• tests/test_batching.py (modified) (5 diffs)

main/waeup.kofa/trunk/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,
+    IExportJobContainer, IExportJob)
 
 class BatchProcessor(grok.GlobalUtility):
@@ -361 +366 @@
                     self.writeFailedRow(
                         failed_writer, string_row,
-                        "Cannot remove: no such entry")
+                        "Cannot remove: no such entry.")
                     continue
                 self.delEntry(row, site)
@@ -498 +503 @@
         """
         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`). It will be named after the
+    exporter used with `.csv` filename extension.
+
+    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()
+    filename = '%s.csv' % exporter_name
+    output_path = os.path.join(output_dir, filename)
+    exporter.export_all(site, filepath=output_path)
+    return output_path
+
+class AsyncExportJob(AsyncJob):
+    """An IJob that exports data to CSV files.
+
+    `AsyncExportJob` instances are regular `AsyncJob` instances with a
+    different constructor API. Instead of a callable to execute, you
+    must pass a `site` and some `exporter_name` to trigger an export.
+
+    The real work is done when an instance of this class is put into a
+    queue. See :mod:`waeup.kofa.async` to learn more about
+    asynchronous jobs.
+
+    The `exporter_name` must be the name under which an ICSVExporter
+    utility was registered with the ZCA.
+
+    The `site` must be a valid site  or ``None``.
+
+    The result of an `AsyncExportJob` is the path to generated CSV
+    file. The file will reside in a temporary directory that should be
+    removed after being used.
+    """
+    grok.implements(IExportJob)
+
+    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.
+    """
+    grok.implements(IExportJobContainer)
+    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)
+        # Make sure that the persisted list is stored in ZODB
+        self.running_exports = PersistentList(self.running_exports)
+        self.running_exports.append((job_id, exporter_name, user_id))
+        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 given entry from the local `running_exports` list and also
+        removes the regarding job via the local job manager.
+
+        `entry` must be 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
+
+    def entry_from_job_id(self, job_id):
+        """Get entry tuple for `job_id`.
+
+        Returns ``None`` if no such entry can be found.
+        """
+        for entry in self.running_exports:
+            if entry[0] == job_id:
+                return entry
+        return None

main/waeup.kofa/trunk/src/waeup/kofa/utils/tests/test_batching.py

@@ -23 +23 @@
 import tempfile
 import unittest
+from zc.async.interfaces import IJob, COMPLETED
 from zope import schema
-from zope.component import provideUtility
+from zope.component import provideUtility, getGlobalSiteManager
 from zope.component.factory import Factory
 from zope.component.hooks import clearSite
@@ -30 +31 @@
 from zope.interface import Interface, implements, verify
 from waeup.kofa.app import University
-from waeup.kofa.interfaces import ICSVExporter, IBatchProcessor
+from waeup.kofa.interfaces import (
+    ICSVExporter, IBatchProcessor, IExportJobContainer, IJobManager,
+    IExportJob)
 from waeup.kofa.testing import FunctionalLayer, FunctionalTestCase
-from waeup.kofa.utils.batching import ExporterBase
+from waeup.kofa.utils.batching import (
+    ExporterBase, BatchProcessor, export_job, AsyncExportJob,
+    ExportJobContainer)
 
 optionflags = (
@@ -71 +76 @@
         self.owner = owner
         self.taxpayer = taxpayer
-#Cave = attrs_to_fields(Cave)
 
 stoneville = dict
 
-from waeup.kofa.utils.batching import BatchProcessor
+SAMPLE_DATA = """name,dinoports,owner,taxpayer
+Barneys Home,2,Barney,1
+Wilmas Asylum,1,Wilma,1
+Freds Dinoburgers,10,Fred,0
+Joeys Drive-in,110,Joey,0
+"""
+
 class CaveProcessor(BatchProcessor):
     util_name = 'caveprocessor'
-    #grok.name(util_name)
     name = 'Cave Processor'
     iface = ICave
@@ -136 +145 @@
         # Provide sample data
         self.newcomers_csv = os.path.join(self.workdir, 'newcomers.csv')
-        open(self.newcomers_csv, 'wb').write(
-            """name,dinoports,owner,taxpayer
-Barneys Home,2,Barney,1
-Wilmas Asylum,1,Wilma,1
-Freds Dinoburgers,10,Fred,0
-Joeys Drive-in,110,Joey,0
-""")
+        open(self.newcomers_csv, 'wb').write(SAMPLE_DATA)
         self.setupLogger()
         self.stoneville = stoneville
@@ -336 +339 @@
         self.assertEqual(result, None)
         return
+
+
+class CaveExporter(ExporterBase):
+    # A minimal fake exporter suitable to be called by export_jobs
+    fields = ('name', 'dinoports', 'owner', 'taxpayer')
+    title = u'Dummy cave exporter'
+
+    def export_all(self, site, filepath=None):
+        if filepath is None:
+            return SAMPLE_DATA
+        open(filepath, 'wb').write(SAMPLE_DATA)
+        return
+
+class ExportJobTests(unittest.TestCase):
+    # Test asynchronous export functionality (simple cases)
+
+    def setUp(self):
+        # register a suitable ICSVExporter as named utility
+        self.exporter = CaveExporter()
+        self.gsm = getGlobalSiteManager()
+        self.gsm.registerUtility(
+            self.exporter, ICSVExporter, name='cave_exporter')
+
+    def tearDown(self):
+        self.gsm.unregisterUtility(self.exporter)
+
+    def test_export_job_func(self):
+        # the export_job func does really export data...
+        result_path = export_job(None, 'cave_exporter')
+        self.assertTrue(os.path.isfile(result_path))
+        contents = open(result_path, 'rb').read()
+        shutil.rmtree(os.path.dirname(result_path))
+        self.assertEqual(contents, SAMPLE_DATA)
+        return
+
+    def test_export_job_interfaces(self):
+        # the AsyncExportJob implements promised interfaces correctly...
+        job = AsyncExportJob(None, None)
+        verify.verifyClass(IJob, AsyncExportJob)
+        verify.verifyObject(IJob, job)
+        verify.verifyClass(IExportJob, AsyncExportJob)
+        verify.verifyObject(IExportJob, job)
+        return
+
+
+class FakeJob(object):
+
+    status = COMPLETED
+    result = None
+
+class FakeJobWithResult(FakeJob):
+
+    def __init__(self):
+        dir_path = tempfile.mkdtemp()
+        self.result = os.path.join(dir_path, 'fake.csv')
+        open(self.result, 'wb').write('a fake result')
+        return
+
+class FakeJobManager(object):
+
+    _jobs = dict()
+    _curr_num = 1
+
+    def get(self, job_id):
+        if job_id == '3':
+            return FakeJob()
+        return self._jobs.get(job_id, None)
+
+    def put(self, job):
+        num = str(self._curr_num)
+        self._jobs[num] = job
+        self._curr_num += 1
+        return num
+
+    def remove(self, job_id, site):
+        if job_id in self._jobs:
+            del self._jobs[job_id]
+        return
+
+class ExportJobContainerTests(unittest.TestCase):
+    # Test ExportJobContainer
+
+    def setUp(self):
+        # register a suitable ICSVExporter as named utility
+        self.exporter = CaveExporter()
+        self.job_manager = FakeJobManager()
+        self.gsm = getGlobalSiteManager()
+        self.gsm.registerUtility(
+            self.exporter, ICSVExporter, name='cave_exporter')
+        self.gsm.registerUtility(
+            self.job_manager, IJobManager)
+
+    def tearDown(self):
+        self.gsm.unregisterUtility(self.exporter)
+        self.gsm.unregisterUtility(self.job_manager, IJobManager)
+
+    def test_export_job_interfaces(self):
+        # the ExportJobContainer implements promised interfaces correctly...
+        container = ExportJobContainer()
+        verify.verifyClass(IExportJobContainer, ExportJobContainer)
+        verify.verifyObject(IExportJobContainer, container)
+        return
+
+    def test_start_export_job(self):
+        # we can start jobs
+        container = ExportJobContainer()
+        container.start_export_job('cave_exporter', 'bob')
+        result = self.job_manager._jobs.values()[0]
+        self.assertTrue(IJob.providedBy(result))
+        self.assertEqual(
+            container.running_exports,
+            [('1', 'cave_exporter', 'bob')]
+            )
+        return
+
+    def test_get_running_export_jobs_all(self):
+        # we can get export jobs of all users
+        container = ExportJobContainer()
+        container.start_export_job('cave_exporter', 'bob')
+        container.start_export_job('cave_exporter', 'alice')
+        result = container.get_running_export_jobs()
+        self.assertEqual(
+            result,
+            [('1', 'cave_exporter', 'bob'),
+             ('2', 'cave_exporter', 'alice')]
+            )
+        return
+
+    def test_get_running_export_jobs_user(self):
+        # we can get the export jobs running for a certain user
+        container = ExportJobContainer()
+        container.start_export_job('cave_exporter', 'bob')
+        container.start_export_job('cave_exporter', 'alice')
+        result1 = container.get_running_export_jobs(user_id='alice')
+        result2 = container.get_running_export_jobs(user_id='foo')
+        self.assertEqual(
+            result1, [('2', 'cave_exporter', 'alice')])
+        self.assertEqual(
+            result2, [])
+        return
+
+    def test_get_running_export_jobs_only_if_exist(self):
+        # we get only jobs that are accessible through the job manager...
+        container = ExportJobContainer()
+        container.start_export_job('cave_exporter', 'bob')
+        container.start_export_job('cave_exporter', 'bob')
+        self.assertTrue(
+            ('2', 'cave_exporter', 'bob') in container.running_exports)
+        # we remove the second entry from job manager
+        del self.job_manager._jobs['2']
+        result = container.get_running_export_jobs(user_id='bob')
+        self.assertEqual(
+            result, [('1', 'cave_exporter', 'bob')])
+        self.assertTrue(
+            ('2', 'cave_exporter', 'bob') not in container.running_exports)
+        return
+
+    def test_get_export_job_status(self):
+        # we can get the stati of jobs...
+        container = ExportJobContainer()
+        container.start_export_job('cave_exporter', 'alice')
+        container.start_export_job('cave_exporter', 'bob')
+        container.start_export_job('cave_exporter', 'bob')
+        result = container.get_export_jobs_status(user_id='bob')
+        # we'll get the raw value, a translation and the title of the
+        # exporter
+        self.assertEqual(
+            result,
+            [('new', u'new', u'Dummy cave exporter'),
+             ('completed', u'completed', u'Dummy cave exporter')]
+            )
+        return
+
+    def test_delete_export_entry(self):
+        # we can remove export entries in local lists and the job
+        # manager as well...
+        container = ExportJobContainer()
+        container.start_export_job('cave_exporter', 'bob')
+        entry = container.running_exports[0]
+        container.delete_export_entry(entry)
+        # both, running_exports list and job manager are empty now
+        self.assertEqual(
+            container.running_exports, [])
+        self.assertEqual(
+            self.job_manager._jobs, {})
+        return
+
+    def test_delete_export_entry_remove_file(self):
+        # any result files of exports are deleted as well
+        container = ExportJobContainer()
+        entry = ('4', 'cave_exporter', 'bob')
+        container.running_exports = [entry]
+        fake_job = FakeJobWithResult()
+        self.job_manager._jobs['4'] = fake_job
+        self.assertTrue(os.path.isfile(fake_job.result))
+        container.delete_export_entry(entry)
+        self.assertTrue(not os.path.exists(fake_job.result))
+        return
+
+    def test_entry_from_job_id(self):
+        # we can get an entry for a job_id if the id exists
+        container = ExportJobContainer()
+        entry = ('4', 'cave_exporter', 'bob')
+        container.running_exports = [entry]
+        fake_job = FakeJobWithResult()
+        self.job_manager._jobs['4'] = fake_job
+        result1 = container.entry_from_job_id(None)
+        result2 = container.entry_from_job_id('4')
+        result3 = container.entry_from_job_id('23')
+        self.assertEqual(result1, None)
+        self.assertEqual(result2, ('4', 'cave_exporter', 'bob'))
+        self.assertEqual(result3, None)
+        return