Source code for waeup.kofa.browser.async

import grok
from zc.async.interfaces import COMPLETED, NEW
from zope.interface import Interface
from waeup.kofa.async import IJob, IJobManager
from waeup.kofa.browser.layout import KofaPage
from waeup.kofa.interfaces import IKofaObject, IDataCenter
from waeup.kofa.interfaces import MessageFactory as _


loadingbar_template = '''
<input id="status_url" type="hidden" name="status_url" value="%s" />
<div id="loadbar">%%s</div>

[docs]class LoadingBar(grok.ViewletManager):'loadingbar')
[docs]class LoadingBarViewlet(grok.Viewlet): grok.context(Interface) grok.viewletmanager(LoadingBar) _running_exports = None
[docs] def getRunningExports(self): """Returns running exports as list of tuples. Only exports triggered by the current user (identified by are returned. Each tuple has the form (<STATUS>, <STATUS_TITLE>, <EXPORTER_NAME>). ``STATUS``: the status as machine readable string (something like ``'completed'``) ``STATUS_TITLE``: status of export as translated string. ``EXPORTER_NAME``: string representing the exporter title used when triggering the export job. """ result = self.context.get_export_jobs_status(self.user_id) return result
[docs] def update(self): self.user_id = self.exports = self.getRunningExports() self.uncompleted = [x for x in self.exports if x[0] != 'completed']
[docs] def render(self): status_url = self.view.url('status') template = loadingbar_template % (status_url,) content = '' return template % content
[docs]class DataCenterJSON(grok.JSON): grok.context(IDataCenter)
[docs] def status(self): """Get status of currently handled export jobs for users. The result is a set of valus useful for JavaScript handlers to update some page handling data center exports. Returns a mapping:: {html=<string>, interval=<int>, reload=<bool>} where `html` is some HTML code that might be shown in a loading bar section. `interval` is the time in microseconds after which the loading bar should be reloaded (by a new JSON call). If the value is zero, then the loading bar should not be renewed and further JSON calls be canceled. `reload` is a hint to tell, whether the complete page should be reloaded or not. Reloading is neccessary, if the export file creation finished and download buttons or similar should appear. """ user_id = jobs = self.context.get_export_jobs_status(user_id) uncompleted = [x for x in jobs if x[0] != 'completed'] result = dict( html='', interval=0, reload=False) if len(uncompleted): html = '<em>%s</em>' % (_("Please wait..."),) result.update(interval=500, html=html) else: if len(jobs): result.update(reload=True) return result
[docs]class JobManagerView(KofaPage): """The main view for the job container. """'index.html') grok.context(IJobManager) grok.template('virtjobscontainerindex') grok.require('waeup.manageJobs')
[docs] def jobs(self): for job_id, job in sorted(, cmp=lambda x, y: cmp(int(x[0]), int(y[0]))): link = self.url(self.context, job_id) removable = job.status in (COMPLETED, NEW) yield dict( job_id = job_id, job = job, link = link, status = job.status, begin_after = job.begin_after, begin_by = job.begin_by, annotations = '%r' % [ (x, y) for x, y in job.annotations.items()], removable = removable, failure = '%r' % getattr(job, 'traceback', None), )
[docs] def update(self, START_NEW=None, REMOVE=None, job_id=None): if REMOVE and job_id: self.context.remove(job_id) self.flash('Removed job "%s"' % job_id) if START_NEW: self.context.start_test_job() #pragma NO COVER self.flash('Started new test job "%s"' % job_id) #pragma NO COVER return
[docs]class JobView(grok.View):'index.html') grok.context(IJob) grok.template('jobindex') grok.require('waeup.manageJobs')
[docs] def percent(self): value = self.context.annotations.get('percent', None) if not value: return value = int(value+0.5) return min(value, 100)
[docs] def update(self): pass