Added support for http callbacks on processing

Sends an HTTP POST request back to an URL given on submission to the API
submit view.

mediagoblin/db/sql/migrations.py

@@ -20,7 +20,8 @@ from sqlalchemy import (MetaData, Table, Column, Boolean, SmallInteger,
                         Integer, Unicode, UnicodeText, DateTime, ForeignKey)
 
 from mediagoblin.db.sql.util import RegisterMigration
-from mediagoblin.db.sql.models import MediaEntry, Collection, User
+from mediagoblin.db.sql.models import MediaEntry, Collection, User, \
+        ProcessingMetaData
 
 MIGRATIONS = {}
 
@@ -101,3 +102,9 @@ def add_mediaentry_collected(db_conn):
     col = Column('collected', Integer, default=0)
     col.create(media_entry)
     db_conn.commit()
+
+
+@RegisterMigration(6, MIGRATIONS)
+def create_processing_metadata_table(db):
+    ProcessingMetaData.__table__.create(db.bind)
+    db.commit()

mediagoblin/db/sql/models.py

@@ -412,9 +412,24 @@ class CollectionItem(Base, CollectionItemMixin):
         return DictReadAttrProxy(self)
 
 
+class ProcessingMetaData(Base):
+    __tablename__ = 'core__processing_metadata'
+
+    id = Column(Integer, primary_key=True)
+    media_entry_id = Column(Integer, ForeignKey(MediaEntry.id), nullable=False,
+            index=True)
+    media_entry = relationship(MediaEntry, backref='processing_metadata')
+    callback_url = Column(Unicode)
+
+    @property
+    def dict_view(self):
+        """A dict like view on this object"""
+        return DictReadAttrProxy(self)
+
+
 MODELS = [
     User, MediaEntry, Tag, MediaTag, MediaComment, Collection, CollectionItem, MediaFile, FileKeynames,
-    MediaAttachmentFile]
+    MediaAttachmentFile, ProcessingMetaData]
 
 
 ######################################################

mediagoblin/plugins/api/views.py

@@ -98,6 +98,12 @@ def post_entry(request):
     # Save now so we have this data before kicking off processing
     entry.save(validate=True)
 
+    if request.POST.get('callback_url'):
+        metadata = request.db.ProcessingMetaData()
+        metadata.media_entry = entry
+        metadata.callback_url = unicode(request.POST['callback_url'])
+        metadata.save()
+
     # Pass off to processing
     #
     # (... don't change entry after this point to avoid race

mediagoblin/processing/task.py

@@ -22,6 +22,7 @@ from mediagoblin import mg_globals as mgg
 from mediagoblin.db.util import ObjectId
 from mediagoblin.media_types import get_media_manager
 from mediagoblin.processing import mark_entry_failed, BaseProcessingFail
+from mediagoblin.tools.processing import json_processing_callback
 
 _log = logging.getLogger(__name__)
 logging.basicConfig()
@@ -58,8 +59,10 @@ class ProcessMedia(Task):
             entry.state = u'processed'
             entry.save()
 
+            json_processing_callback(entry)
         except BaseProcessingFail as exc:
             mark_entry_failed(entry._id, exc)
+            json_processing_callback(entry)
             return
 
         except ImportError as exc:
@@ -70,6 +73,7 @@ class ProcessMedia(Task):
                     exc))
 
             mark_entry_failed(entry._id, exc)
+            json_processing_callback(entry)
 
         except Exception as exc:
             _log.error('An unhandled exception was raised while'
@@ -77,6 +81,7 @@ class ProcessMedia(Task):
                         entry))
 
             mark_entry_failed(entry._id, exc)
+            json_processing_callback(entry)
             raise
 
     def on_failure(self, exc, task_id, args, kwargs, einfo):
@@ -90,3 +95,6 @@ class ProcessMedia(Task):
         """
         entry_id = args[0]
         mark_entry_failed(entry_id, exc)
+
+        entry = mgg.database.MediaEntry.query.filter_by(id=entry_id)
+        json_processing_callback(entry)

mediagoblin/tools/processing.py

@@ -0,0 +1,73 @@
+# GNU MediaGoblin -- federated, autonomous media hosting
+# Copyright (C) 2011, 2012 MediaGoblin contributors.  See AUTHORS.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import logging
+import json
+
+from urllib2 import urlopen, Request
+from urllib import urlencode
+
+_log = logging.getLogger(__name__)
+
+
+def create_post_request(url, data, **kw):
+    '''
+    Issue a HTTP POST request.
+
+    Args:
+        url: The URL to which the POST request should be issued
+        data: The data to be send in the body of the request
+        **kw:
+        data_parser: The parser function that is used to parse the `data`
+            argument
+    '''
+    data_parser = kw.get('data_parser', urlencode)
+    headers = kw.get('headers', {})
+
+    return Request(url, data_parser(data), headers=headers)
+
+
+def json_processing_callback(entry):
+    '''
+    Send an HTTP post to the registered callback url, if any.
+    '''
+    if not entry.processing_metadata:
+        _log.debug('No processing callback for {0}'.format(entry))
+        return
+
+    url = entry.processing_metadata[0].callback_url
+
+    _log.debug('Sending processing callback for {0} ({1})'.format(
+        entry,
+        url))
+
+    headers = {
+            'Content-Type': 'application/json'}
+
+    data = {
+            'id': entry.id,
+            'state': entry.state}
+
+    request = create_post_request(
+            url,
+            data,
+            headers=headers,
+            data_parser=json.dumps)
+
+    urlopen(request)
+    _log.debug('Processing callback for {0} sent'.format(entry))
+
+    return True