1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
|
# Copyright 2011 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Job for merging translations."""
__metaclass__ = type
__all__ = [
'POFileStatsJob',
]
import logging
from storm.locals import (
And,
Int,
Reference,
)
from zope.component import getUtility
from zope.interface import (
classProvides,
implements,
)
from lp.services.database.stormbase import StormBase
from lp.services.job.interfaces.job import IRunnableJob
from lp.services.job.model.job import Job
from lp.services.job.runner import BaseRunnableJob
from lp.services.webapp.interfaces import (
DEFAULT_FLAVOR,
IStoreSelector,
MAIN_STORE,
)
from lp.translations.interfaces.pofilestatsjob import IPOFileStatsJobSource
from lp.translations.model.pofile import POFile
class POFileStatsJob(StormBase, BaseRunnableJob):
"""The details for a POFile status update job."""
__storm_table__ = 'POFileStatsJob'
# Instances of this class are runnable jobs.
implements(IRunnableJob)
# Oddly, BaseRunnableJob inherits from BaseRunnableJobSource so this class
# is both the factory for jobs (the "implements", above) and the source
# for runnable jobs (not the constructor of the job source, the class
# provides the IJobSource interface itself).
classProvides(IPOFileStatsJobSource)
# The Job table contains core job details.
job_id = Int('job', primary=True)
job = Reference(job_id, Job.id)
# This is the POFile which needs its statistics updated.
pofile_id = Int('pofile')
pofile = Reference(pofile_id, POFile.id)
def __init__(self, pofile):
self.job = Job()
self.pofile = pofile
super(POFileStatsJob, self).__init__()
def getOperationDescription(self):
"""See `IRunnableJob`."""
return 'updating POFile statistics'
def run(self):
"""See `IRunnableJob`."""
logger = logging.getLogger()
logger.info('Updating statistics for %s' % self.pofile.title)
self.pofile.updateStatistics()
@staticmethod
def iterReady():
"""See `IJobSource`."""
store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
return store.find((POFileStatsJob),
And(POFileStatsJob.job == Job.id,
Job.id.is_in(Job.ready_jobs)))
def schedule(pofile):
"""Schedule a job to update a POFile's stats."""
return POFileStatsJob(pofile)
|