[#2316] add resource status api

ckan/ckan_nose_plugin.py

@@ -30,7 +30,6 @@ def startContext(self, ctx):
             # init_db is run at the start of every class because
             # when you use an in-memory sqlite db, it appears that
             # the db is destroyed after every test when you Session.Remove().
-            model.repo.init_db()
 
             ## This is to make sure the configuration is run again.
             ## Plugins use configure to make their own tables and they
@@ -40,6 +39,7 @@ def startContext(self, ctx):
             for plugin in PluginImplementations(IConfigurable):
                 plugin.configure(config)
 
+            model.repo.init_db()
 
     def options(self, parser, env):
         parser.add_option(

ckan/config/environment.py

@@ -151,9 +151,8 @@ def template_loaded(template):
     ckan_db = os.environ.get('CKAN_DB') 
 
     if ckan_db:
-        engine = sqlalchemy.create_engine(ckan_db)
-    else:
-        engine = sqlalchemy.engine_from_config(config, 'sqlalchemy.')
+        config['sqlalchemy.url'] = ckan_db
+    engine = sqlalchemy.engine_from_config(config, 'sqlalchemy.')
 
     if not model.meta.engine:
         model.init_model(engine)

ckan/lib/celery_app.py

@@ -1,5 +1,6 @@
 import ConfigParser
 import os
+from pylons import config as pylons_config
 from pkg_resources import iter_entry_points
 #from celery.loaders.base import BaseLoader
 
@@ -12,16 +13,22 @@
 config = ConfigParser.ConfigParser()
 
 config_file = os.environ.get('CKAN_CONFIG')
+
 if not config_file:
     config_file =  os.path.join(
         os.path.dirname(os.path.abspath(__file__)), '../../development.ini')
 config.read(config_file)
 
 
+sqlalchemy_url = pylons_config.get('sqlalchemy.url')
+if not sqlalchemy_url:
+    sqlalchemy_url = config.get('app:main', 'sqlalchemy.url')
+
+
 default_config = dict( 
     BROKER_BACKEND = 'sqlalchemy',
-    BROKER_HOST = config.get('app:main', 'sqlalchemy.url'),
-    CELERY_RESULT_DBURI = config.get('app:main', 'sqlalchemy.url'),
+    BROKER_HOST = sqlalchemy_url,
+    CELERY_RESULT_DBURI = sqlalchemy_url,
     CELERY_RESULT_BACKEND = 'database',
     CELERY_RESULT_SERIALIZER = 'json',
     CELERY_TASK_SERIALIZER = 'json',

ckan/logic/__init__.py

@@ -39,6 +39,7 @@ class NotAuthorized(ActionError):
 class ParameterError(ActionError):
     pass
 
+
 class ValidationError(ParameterError):
     def __init__(self, error_dict, error_summary=None, extra_msg=None):
         self.error_dict = error_dict
@@ -224,3 +225,29 @@ def get_action(action):
     _actions.update(fetched_actions)
     return _actions.get(action)
 
+def get_or_bust(data_dict, keys):
+    '''Try and get values from dictionary and if they are not there
+    raise a validataion error.
+
+    data_dict: a dictionary
+    keys: either a single string key in which case will return a single value,
+    or a iterable which will return a tuple for unpacking purposes.
+
+    e.g single_value = get_or_bust(data_dict, 'a_key')
+        value_1, value_2 = get_or_bust(data_dict, ['key1', 'key2'])
+    '''
+    values = []
+    errors = {}
+
+    if isinstance(keys, basestring):
+        keys = [keys]
+    for key in keys:
+        value = data_dict.get(key)
+        if not value:
+            errors[key] = _('Missing value')
+        values.append(value)
+    if errors:
+        raise ValidationError(errors)
+    if len(values) == 1:
+        return values[0]
+    return tuple(values)

ckan/logic/action/get.py

@@ -6,7 +6,7 @@
 import webhelpers.html
 from sqlalchemy.sql import select
 from sqlalchemy.orm import aliased
-from sqlalchemy import or_, and_, func, desc, case
+from sqlalchemy import or_, and_, func, desc, case, text
 
 import ckan
 import ckan.authz
@@ -31,6 +31,7 @@
 check_access = logic.check_access
 NotFound = logic.NotFound
 ValidationError = logic.ValidationError
+get_or_bust = logic.get_or_bust
 
 def _package_list_with_resources(context, package_revision_list):
     package_list = []
@@ -417,9 +418,27 @@ def resource_show(context, data_dict):
         raise NotFound
 
     check_access('resource_show', context, data_dict)
-
     return model_dictize.resource_dictize(resource, context)
 
+def resource_status_show(context, data_dict):
+
+    model = context['model']
+    id = get_or_bust(data_dict, 'id')
+
+    check_access('resource_status_show', context, data_dict)
+
+    # needs to be text query as celery tables are not in our model
+    q = text("""select status, date_done, traceback, task_status.* 
+                from task_status left join celery_taskmeta 
+                on task_status.value = celery_taskmeta.task_id and key = 'celery_task_id' 
+                where entity_id = :entity_id """)
+
+    result = model.Session.connection().execute(q, entity_id=id)
+    result_list = [table_dictize(row, context) for row in result]
+
+    return result_list
+
+
 def revision_show(context, data_dict):
     model = context['model']
     api = context.get('api_version')

ckan/logic/auth/get.py

@@ -154,6 +154,9 @@ def format_autocomplete(context, data_dict):
 def task_status_show(context, data_dict):
     return {'success': True}
 
+def resource_status_show(context, data_dict):
+    return {'success': True}
+
 ## Modifications for rest api
 
 def package_show_rest(context, data_dict):

ckan/logic/auth/publisher/get.py

@@ -160,6 +160,9 @@ def format_autocomplete(context, data_dict):
 def task_status_show(context, data_dict):
     return {'success': True}
 
+def resource_status_show(context, data_dict):
+    return {'success': True}
+
 ## Modifications for rest api
 
 def package_show_rest(context, data_dict):

ckan/model/__init__.py

@@ -79,6 +79,16 @@ def init_db(self):
         else:
             if not self.tables_created_and_initialised:
                 self.upgrade_db()
+                ## make sure celery tables are made as celery only makes them after
+                ## adding a task
+                try:
+                    import ckan.lib.celery_app as celery_app
+                    backend = celery_app.celery.backend
+                    ##This creates the database tables as a side effect, can not see another way
+                    ##to make tables unless you actually create a task.
+                    celery_result_session = backend.ResultSession()
+                except ImportError:
+                    pass
                 self.init_configuration_data()
                 self.tables_created_and_initialised = True
 

ckan/tests/logic/test_action.py

@@ -1542,6 +1542,27 @@ def test_38_user_role_bulk_update(self):
                        {'domain_object': anna.id})
         assert_equal(results['roles'], roles_after['roles'])
 
+    def test_40_task_resource_status(self):
+
+        import ckan.lib.celery_app as celery_app
+        backend = celery_app.celery.backend
+        ##This creates the database tables as a side effect, can not see another way
+        ##to make tables unless you actually create a task.
+        celery_result_session = backend.ResultSession()
+
+        ## need to do inserts as setting up an embedded celery is too much for these tests
+        model.Session.connection().execute(
+            '''INSERT INTO task_status (id, entity_id, entity_type, task_type, key, value, state, error, last_updated) VALUES ('5753adae-cd0d-4327-915d-edd832d1c9a3', '749cdcf2-3fc8-44ae-aed0-5eff8cc5032c', 'resource', 'qa', 'celery_task_id', '51f2105d-85b1-4393-b821-ac11475919d9', NULL, '', '2012-04-20 21:32:45.553986');
+               INSERT INTO celery_taskmeta (id, task_id, status, result, date_done, traceback) VALUES (2, '51f2105d-85b1-4393-b821-ac11475919d9', 'FAILURE', '52e', '2012-04-20 21:33:01.622557', 'Traceback')'''
+        )
+        model.Session.commit()
+        res = self.app.post('/api/action/resource_status_show', 
+                            params=json.dumps({'id': '749cdcf2-3fc8-44ae-aed0-5eff8cc5032c'}),
+                            status=200)
+
+        assert json.loads(res.body) == {"help": None, "success": True, "result": [{"status": "FAILURE", "entity_id": "749cdcf2-3fc8-44ae-aed0-5eff8cc5032c", "task_type": "qa", "last_updated": "2012-04-20T21:32:45.553986", "date_done": "2012-04-20T21:33:01.622557", "entity_type": "resource", "traceback": "Traceback", "value": "51f2105d-85b1-4393-b821-ac11475919d9", "state": None, "key": "celery_task_id", "error": "", "id": "5753adae-cd0d-4327-915d-edd832d1c9a3"}]}
+
+
 class TestActionTermTranslation(WsgiAppCase):
 
     @classmethod

doc/apiv3.rst

@@ -78,6 +78,9 @@ package_show_rest                      id
 group_show_rest                        id
 tag_show_rest                          id
 vocabulary_show                        id
+task_status_show                       id
+task_status_show                       entity_id, task_type, key 
+resource_status_show                   id
 package_autocomplete                   q
 tag_autocomplete                       q, fields, offset, limit, vocabulary_id
 format_autocomplete                    q, limit