Removed unecessary use of %.
#!/usr/bin/python2.5
#
# Copyright 2009 the Melange authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Cron jobs.
"""
__authors__ = [
'"Sverre Rabbelier" <sverre@rabbelier.nl>',
'"Lennard de Rijk" <ljvderijk@gmail.com>',
]
import logging
from google.appengine.ext import db
from google.appengine.runtime import DeadlineExceededError
from soc.cron import student_proposal_mailer
from soc.cron import unique_user_id_adder
from soc.models.job import Job
class Error(Exception):
"""Base class for all exceptions raised by this module.
"""
pass
class FatalJobError(Error):
"""Class for all errors that lead to immediate job abortion.
"""
pass
class Handler(object):
"""A handler that dispatches a cron job.
The tasks that are mapped into tasks will be called when a worker
has claimed the job. However, there is no guarantee as to how long
the task will be allowed to run. If an Exception is raised the task
is automatically rescheduled for execution.
"""
def __init__(self):
"""Constructs a new Handler with all known jobs set.
"""
# pylint: disable-msg=C0103
self.OUT_OF_TIME = 0
self.ALREADY_CLAIMED = 1
self.SUCCESS = 2
self.ABORTED = 3
self.ERRORED = 4
self.tasks = {}
self.tasks['setupStudentProposalMailing'] = \
student_proposal_mailer.setupStudentProposalMailing
self.tasks['sendStudentProposalMail'] = \
student_proposal_mailer.sendStudentProposalMail
self.tasks['setupUniqueUserIdAdder'] = \
unique_user_id_adder.setupUniqueUserIdAdder
self.tasks['addUniqueUserIds'] = \
unique_user_id_adder.addUniqueUserIds
def claimJob(self, job_key):
"""A transaction to claim a job.
The transaction is rolled back if the status is not 'waiting'.
"""
job = Job.get_by_id(job_key)
if job.status != 'waiting':
raise db.Rollback()
job.status = 'started'
# pylint: disable-msg=E1103
if job.put():
return job
else:
return None
def timeoutJob(self, job):
"""Timeout a job.
If a job has timed out more than 50 times, the job is aborted.
"""
job.timeouts += 1
if job.timeouts > 50:
job.status = 'aborted'
else:
job.status = 'waiting'
job.put()
job_id = job.key().id()
logging.debug("job %d now timeout %d time(s)" % (job_id, job.timeouts))
def failJob(self, job):
"""Fail a job.
If the job has failed more than 5 times, the job is aborted.
"""
job.errors += 1
if job.errors > 5:
job.status = 'aborted'
else:
job.status = 'waiting'
job.put()
job_id = job.key().id()
logging.warning("job %d now failed %d time(s)" % (job_id, job.errors))
def finishJob(self, job):
"""Finish a job.
"""
job.status = 'finished'
job.put()
def abortJob(self, job):
"""Abort a job.
"""
job.status = 'aborted'
job.put()
def handle(self, job_key):
"""Handle one job.
Returns: one of the following status codes:
self.OUT_OF_TIME: returned when a DeadlineExceededError is raised
self.ALREADY_CLAIMED: if job.status is not 'waiting'
self.SUCCESS: if the job.status has been set to 'succes'
self.ABORTED: if the job.status has been set to 'aborted'
self.ERRORED: if the job encountered an error
"""
job = None
try:
job = db.run_in_transaction(self.claimJob, job_key)
if not job:
# someone already claimed the job
return self.ALREADY_CLAIMED
if job.task_name not in self.tasks:
logging.error("Unknown job %s" % job.task_name)
db.run_in_transaction(self.abortJob, job_key)
return self.ABORTED
task = self.tasks[job.task_name]
# execute the actual job
task(job)
self.finishJob(job)
return self.SUCCESS
except DeadlineExceededError, exception:
if job:
self.timeoutJob(job)
return self.OUT_OF_TIME
except FatalJobError, exception:
logging.exception(exception)
if job:
self.abortJob(job)
return self.ABORTED
except Exception, exception:
logging.exception(exception)
if job:
self.failJob(job)
return self.ERRORED
def iterate(self, jobs, retry_jobs):
"""Trivial iterator that iterates over jobs then retry_jobs
"""
for job in jobs:
yield job
while retry_jobs:
yield retry_jobs[0]
handler = Handler()