Merge branch 'feature-1298-activities-table' into feature-1515-activi…

…ty-streams

ckan/lib/activity.py

@@ -0,0 +1,119 @@
+from sqlalchemy.orm.session import SessionExtension
+import logging
+logger = logging.getLogger(__name__)
+
+def activity_stream_item(obj, activity_type, revision_id):
+    try:
+        return obj.activity_stream_item(activity_type, revision_id)
+    except (AttributeError, TypeError):
+        logger.debug("Object did not have a suitable "
+            "activity_stream_item() method, it must not be a package.")
+        return None
+
+def activity_stream_detail(obj, activity_id, activity_type):
+    try:
+        return obj.activity_stream_detail(activity_id, activity_type)
+    except (AttributeError, TypeError):
+        logger.debug("Object did not have a suitable  "
+            "activity_stream_detail() method.")
+        return None
+
+class DatasetActivitySessionExtension(SessionExtension):
+
+    def before_commit(self, session):
+
+        session.flush()
+
+        try:
+            obj_cache = session._object_cache
+            revision = session.revision
+        except AttributeError:
+            return
+
+        # The top-level objects that we will append to the activity table. The
+        # keys here are package IDs, and the values are model.activity:Activity
+        # objects.
+        activities = {}
+
+        # The second-level objects that we will append to the activity_detail
+        # table. Each row in the activity table has zero or more related rows
+        # in the activity_detail table. The keys here are activity IDs, and the
+        # values are lists of model.activity:ActivityDetail objects.
+        activity_details = {}
+
+        # Log new packages first to prevent them from getting incorrectly
+        # logged as changed packages.
+        logger.debug("Looking for new packages...")
+        for obj in obj_cache['new']:
+            logger.debug("Looking at object %s" % obj)
+            activity = activity_stream_item(obj, 'new', revision.id)
+            if activity is None:
+                continue
+            # If the object returns an activity stream item we know that the
+            # object is a package.
+            logger.debug("Looks like this object is a package")
+            logger.debug("activity: %s" % activity)
+            activities[obj.id] = activity
+
+            activity_detail = activity_stream_detail(obj, activity.id, "new")
+            if activity_detail is not None:
+                logger.debug("activity_detail: %s" % activity_detail)
+                activity_details[activity.id] = [activity_detail]
+
+        # Now process other objects.
+        logger.debug("Looking for other objects...")
+        for activity_type in ('new', 'changed', 'deleted'):
+            objects = obj_cache[activity_type]
+            for obj in objects:
+                logger.debug("Looking at %s object %s" % (activity_type, obj))
+                if activity_type == "new" and obj.id in activities:
+                    logger.debug("This object was already logged as a new "
+                            "package")
+                    continue
+
+                try:
+                    related_packages = obj.related_packages()
+                    logger.debug("related_packages: %s" % related_packages)
+                except (AttributeError, TypeError):
+                    logger.debug("Object did not have a suitable "
+                            "related_packages() method, skipping it.")
+                    continue
+
+                for package in related_packages:
+                    if package is None: continue
+
+                    if package.id in activities:
+                        activity = activities[package.id]
+                    else:
+                        activity = activity_stream_item(package, "changed",
+                                revision.id)
+                        activities[package.id] = activity
+                    assert activity is not None
+                    logger.debug("activity: %s" % activity)
+
+                    activity_detail = activity_stream_detail(obj, activity.id,
+                            activity_type)
+                    logger.debug("activity_detail: %s" % activity_detail)
+                    if activity_detail is not None:
+                        if activity_details.has_key(activity.id):
+                            activity_details[activity.id].append(
+                                    activity_detail)
+                        else:
+                            activity_details[activity.id] =  [activity_detail]
+
+        for key, activity in activities.items():
+            logger.debug("Emitting activity: %s %s"
+                    % (activity.id, activity.activity_type))
+            session.add(activity)
+
+        session.flush()
+
+        for key, activity_detail_list in activity_details.items():
+            for activity_detail_obj in activity_detail_list:
+                logger.debug("Emitting activity detail: %s %s %s"
+                        % (activity_detail_obj.activity_id,
+                            activity_detail_obj.activity_type,
+                            activity_detail_obj.object_type))
+                session.add(activity_detail_obj)
+
+        session.flush()

ckan/lib/dictization/model_save.py

@@ -46,6 +46,9 @@ def resource_dict_save(res_dict, context):
     return obj
 
 def package_resource_list_save(res_dicts, package, context):
+    allow_partial_update = context.get("allow_partial_update", False)
+    if not res_dicts and allow_partial_update:
+        return
 
     pending = context.get('pending')
 

ckan/model/activity.py

@@ -0,0 +1,60 @@
+import datetime
+
+from sqlalchemy import orm
+
+from meta import *
+from types import make_uuid
+from core import *
+from package import *
+
+__all__ = ['Activity', 'activity_table', 
+           'ActivityDetail', 'activity_detail_table',
+           ]
+
+activity_table = Table(
+    'activity', metadata,
+    Column('id', types.UnicodeText, primary_key=True, default=make_uuid),
+    Column('timestamp', types.DateTime),
+    Column('user_id', types.UnicodeText),
+    Column('object_id', types.UnicodeText),
+    Column('revision_id', types.UnicodeText),
+    Column('activity_type', types.UnicodeText),
+    Column('data', types.UnicodeText),
+    )
+
+activity_detail_table = Table(
+    'activity_detail', metadata,
+    Column('id', types.UnicodeText, primary_key=True, default=make_uuid),
+    Column('activity_id', types.UnicodeText, ForeignKey('activity.id')),
+    Column('object_id', types.UnicodeText),
+    Column('object_type', types.UnicodeText),
+    Column('activity_type', types.UnicodeText),
+    Column('data', types.UnicodeText),
+    )
+
+class Activity(DomainObject):
+
+    def __init__(self, user_id, object_id, revision_id, activity_type, data):
+        self.id = make_uuid()
+        self.timestamp = datetime.datetime.now()
+        self.user_id = user_id
+        self.object_id = object_id
+        self.revision_id = revision_id
+        self.activity_type = activity_type
+        self.data = data
+
+mapper(Activity, activity_table)
+
+class ActivityDetail(DomainObject):
+
+    def __init__(self, activity_id, object_id, object_type, activity_type,
+            data):
+        self.activity_id = activity_id
+        self.object_id = object_id
+        self.object_type = object_type
+        self.activity_type = activity_type
+        self.data = data
+
+mapper(ActivityDetail, activity_detail_table, properties = {
+    'activity':orm.relation ( Activity, backref=orm.backref('activity_detail'))
+    })

ckan/model/meta.py

@@ -15,6 +15,8 @@
 
 from ckan.model import extension
 
+from ckan.lib.activity import DatasetActivitySessionExtension
+
 class CkanSessionExtension(SessionExtension):
 
     def before_flush(self, session, flush_context, instances):
@@ -96,15 +98,17 @@ def after_rollback(self, session):
         autoflush=False,
         transactional=True,
         extension=[CkanSessionExtension(),
-                   extension.PluginSessionExtension()],
+                   extension.PluginSessionExtension(),
+                   DatasetActivitySessionExtension()],
         ))
 else:
     Session = scoped_session(sessionmaker(
         autoflush=False,
         autocommit=False,
         expire_on_commit=False,
         extension=[CkanSessionExtension(),
-                   extension.PluginSessionExtension()],
+                   extension.PluginSessionExtension(),
+                   DatasetActivitySessionExtension()],
         ))
 
 #mapper = Session.mapper

ckan/model/package.py

@@ -1,11 +1,13 @@
 import datetime
 from time import gmtime
 from calendar import timegm
+import logging
+logger = logging.getLogger(__name__)
 
 from sqlalchemy.sql import select, and_, union, expression, or_
 from sqlalchemy.orm import eagerload_all
 from sqlalchemy import types, Column, Table
-from pylons import config
+from pylons import config, session, c, request
 from meta import metadata, Session
 import vdm.sqlalchemy
 
@@ -14,6 +16,7 @@
 from license import License, LicenseRegister
 from domain_object import DomainObject
 import ckan.misc
+from activity import Activity, ActivityDetail
 
 __all__ = ['Package', 'package_table', 'package_revision_table',
            'PACKAGE_NAME_MAX_LENGTH', 'PACKAGE_NAME_MIN_LENGTH',
@@ -50,7 +53,7 @@
 class Package(vdm.sqlalchemy.RevisionedObjectMixin,
         vdm.sqlalchemy.StatefulObjectMixin,
         DomainObject):
-    
+
     text_search_fields = ['name', 'title']
 
     def __init__(self, **kw):
@@ -534,3 +537,31 @@ def get_fields(core_only=False, fields_to_ignore=None):
 
         return fields
 
+    def activity_stream_item(self, activity_type, revision_id):
+        try:
+            user_id = c.user_obj.id
+        except TypeError:
+            # Cannot access user ID through Pylons context, try to get their IP
+            # address instead.
+            try:
+                user_id = request.environ.get('REMOTE_ADDR', 'Unknown IP Address')
+            except TypeError:
+                # Cannot access user IP address through Pylons request,
+                # fallback on a default value.
+                user_id = 'Unknown IP Address'
+        logger.debug("user_id: %s" % user_id)
+        assert activity_type in ("new", "changed", "deleted"), \
+            str(activity_type)
+        if activity_type == "new":
+            return Activity(user_id, self.id, revision_id, "new package",
+                None)
+        elif activity_type == "changed":
+            return Activity(user_id, self.id, revision_id, "changed package",
+                None)
+        elif activity_type == "deleted":
+            return Activity(user_id, self.id, revision_id, "deleted package",
+                None)
+
+    def activity_stream_detail(self, activity_id, activity_type):
+        return ActivityDetail(activity_id, self.id, u"Package", activity_type, 
+            data=None)

ckan/model/resource.py

@@ -9,6 +9,7 @@
 from core import *
 from package import *
 from ckan.model import extension
+from ckan.model.activity import ActivityDetail
 
 __all__ = ['Resource', 'resource_table', 
            'ResourceGroup', 'resource_group_table',
@@ -139,6 +140,9 @@ def get_extra_columns(cls):
     def related_packages(self):
         return [self.resource_group.package]
 
+    def activity_stream_detail(self, activity_id, activity_type):
+        return ActivityDetail(activity_id, self.id, u"Resource", activity_type,
+                None)
 
 class ResourceGroup(vdm.sqlalchemy.RevisionedObjectMixin,
                vdm.sqlalchemy.StatefulObjectMixin,