From bf2dafd1a04ef8050ebf08bb512862a1592998c0 Mon Sep 17 00:00:00 2001 From: Sebastian Spaeth Date: Thu, 20 Dec 2012 13:42:37 +0100 Subject: Tweak Celery Task - Make sure Exceptions are pickleable (not sure if this was not the case but this is the pattern as documented in the celery docs. - Don't create a task_id in the GMG code, but save the one implicitely created by celery. - Don't create a task-id directory per upload. Just store queued uploads in a single directory (this is the most controversial change and might need discussion!!!) Signed-off-by: Sebastian Spaeth --- mediagoblin/processing/task.py | 22 ++++++++++------------ 1 file changed, 10 insertions(+), 12 deletions(-) (limited to 'mediagoblin/processing/task.py') diff --git a/mediagoblin/processing/task.py b/mediagoblin/processing/task.py index 9af192ed..550906d0 100644 --- a/mediagoblin/processing/task.py +++ b/mediagoblin/processing/task.py @@ -18,11 +18,13 @@ import logging import urllib import urllib2 -from celery import registry, task +#TODO: newer celeries use from celery import Task. Change when we upgrade +from celery.task import Task +from celery.registry import tasks from mediagoblin import mg_globals as mgg -from mediagoblin.db.models import MediaEntry -from . import mark_entry_failed, BaseProcessingFail, ProcessingState +from mediagoblin.db.sql.models import MediaEntry +from mediagoblin.processing import mark_entry_failed, BaseProcessingFail from mediagoblin.tools.processing import json_processing_callback _log = logging.getLogger(__name__) @@ -63,12 +65,10 @@ def handle_push_urls(feed_url): ################################ # Media processing initial steps ################################ +class ProcessMedia(Task): + track_started=True -class ProcessMedia(task.Task): - """ - Pass this entry off for processing. - """ - def run(self, media_id, feed_url): + def run(self, media_id): """ Pass the media entry off to the appropriate processing function (for now just process_image...) @@ -81,8 +81,8 @@ class ProcessMedia(task.Task): # Try to process, and handle expected errors. try: entry.state = u'processing' + entry.queued_task_id = self.request.id entry.save() - _log.debug('Processing {0}'.format(entry)) proc_state = ProcessingState(entry) @@ -140,6 +140,4 @@ class ProcessMedia(task.Task): entry = mgg.database.MediaEntry.query.filter_by(id=entry_id).first() json_processing_callback(entry) -# Register the task -process_media = registry.tasks[ProcessMedia.name] - +tasks.register(ProcessMedia) -- cgit v1.2.3