Mostly done with major update to entity usage

Author: briehl

README.md

@@ -30,16 +30,38 @@ make test
 (see [the API design doc](design/api.md) for details, also see below.)
 ### Data Structures
 
+**Entity**
+An Entity structure defines, loosely, an entity related to a notification. This will make more sense below in the Notification object, but Entities are all of the actors in the notifications, the objects of each notification, users who receive the notification, and targets of the notification. For example, in the notification defined by the phrase:  
+```
+User "wjriehl" invited user "some_other_user" to join the group "Bill's Fancy Group".
+```
+There are 3 entities:  
+1. wjriehl is an entity of type user
+2. some_other_user is an entity of type user
+3. Bill's Fancy Group is an entity of type group
+These are used as a vocabulary to strictly control what's meant in the notification structure.
+
+The types currently supported are:
+* user - id should be a user id
+* group - id should be a group id
+* workspace - id should be a workspace id
+* narrative - id should be a workspace id (maybe an UPA if needed)
+* job - id should be a job id string
+{
+    "id": string - an id for this notification, defined by context,
+    "type": string - what type of entity this is. Allowed values below.
+    "name": string (optional) - the full name of this entity (not stored, as its possible it could change)
+}
+```
+
 **Notification**
 ```
 {
     "id": string - a unique id for the notification,
-    "actor": string - the id of the actor who triggered the notification,
-    "actor_name": string - the real name (whether group or user) of the actor,
-    "actor_type": string - (one of group, user) - type of the actor,
+    "actor": Entity - the Entity who triggered the notification (mostly a user),
     "verb": string - the action represented by this notification (see list of verbs below),
-    "object": string - the object of the notification,
-    "target": list - the target(s) of the notification,
+    "object": Entity - the object of the notification,
+    "target": list(Entity) - the target(s) of the notification,
     "source": string - the source service that created the notification,
     "level": string, one of alert, error, warning, request,
     "seen": boolean, if true, then this has been seen before,
@@ -190,7 +212,7 @@ This is meant for services to debug their use of external keys. A service that c
 ### Create a new notification
 Only services (i.e. those Authorization tokens with type=Service, as told by the Auth service) can use this endpoint to create a new notification. This requires the body to be a JSON structure with the following available keys (pretty similar to the Notification structure above):
 * `source` - required, this is the source service of the request.
-* `actor` - required, should be a kbase username
+* `actor` - required, should be a valid Entity
 * `object` - required, the object of the notice (the workspace being shared, the group being invited to, etc.)
 * `verb` - required, the action implied by this notice. Currently allowed verbs are:
     * invite / invited
@@ -207,10 +229,11 @@ Only services (i.e. those Authorization tokens with type=Service, as told by the
     * warning
     * error
     * request
-* `target` - (*TODO: update this field*) - currently required, this is a list of user ids that are affected by this notification; and it is also the list of users who see the notification.
+* `target` - optional, but if present should be a list of Entities - the targets of the notification
 * `expires` - optional, an expiration date for the notification in number of milliseconds since the epoch. Default is 30 days after creation.
 * `external_key` - optional, a string that can be used to look up notifications from a service.
 * `context` - optional, a key-value pair structure that can have some semantic meaning for the notification. "Special" keys are `text` - which is used to generate the viewed text in the browser (omitting this will autogenerate the text from the other attributes), and `link` - a URL used to craft a hyperlink in the browser.
+* `users` - optional, a list of Entities that should receive the notification (limited to users and groups). This list will be automatically augmented by the service if necessary. E.g. if a workspace is the object of the notification, then the workspace admins will be notified.
 
 **Usage:**
 * Path: `/api/V1/notification`
@@ -224,20 +247,29 @@ Only services (i.e. those Authorization tokens with type=Service, as told by the
 ```python
 import requests
 note = {
-    "actor": "wjriehl",
+    "actor": {
+        "id": "wjriehl",
+        "type": "user"
+    },
     "source": "workspace",
     "verb": "shared",
-    "object": "30000",
-    "target": ["gaprice"],
+    "object": {
+        "id": "30000",
+        "type": "workspace"
+    },
+    "target": [{
+        "id": "gaprice",
+        "type": "user"
+    }],
     "context": {
-        "text": "User wjriehl shared workspace 30000 with user gaprice."  # this can also be auto-generated
+        "text": "User wjriehl shared workspace 30000 with user gaprice."  # this can also be auto-generated by the UI.
     }
 }
 r = requests.post("https://<service_url>/api/V1/notification", json=note, headers={"Authorization": auth_token})
 ```
 would return:
 ```python
-{"id": "some-uuid-for-the-notification"}
+{"id": "some-unique-id-for-the-notification"}
 ```
 
 ### Mark notifications as seen

TODO.md

@@ -2,6 +2,11 @@
 * Integrate with Groups
     * Make Groups service API
     * Include groups as an actor, target, object
+        * Needs full name lookup
+        * Needs separation between user/group for each type
+        * Probably need to separate target/object from just a list of strings into a list of objects
+            * each object = id, type
+        
     * Return groups in the feeds lookup
         * new "group" key
 * Include total number of unexpired notifications for a user

feeds/activity/notification.py

@@ -3,20 +3,26 @@
 import json
 from ..util import epoch_ms
 from .. import verbs
-from ..actor import validate_actor
 from .. import notification_level
 from feeds.exceptions import (
     InvalidExpirationError,
     InvalidNotificationError
 )
 import datetime
 from feeds.config import get_config
+from feeds.entity import Entity
+from typing import (
+    List,
+    TypeVar
+)
+N = TypeVar('N', bound='Notification')
 
 
 class Notification(BaseActivity):
-    def __init__(self, actor: str, verb, note_object: str, source: str, actor_name: str=None,
-                 actor_type: str='user', level='alert', target: list=None, context: dict=None,
-                 expires: int=None, external_key: str=None, seen: bool=False, users: list=None):
+    def __init__(self, actor: Entity, verb: str, note_object: Entity, source: str,
+                 level='alert', target: List[Entity]=None, context: dict=None,
+                 expires: int=None, external_key: str=None, seen: bool=False,
+                 users: List[Entity]=None):
         """
         A notification is roughly of this form:
             actor, verb, object, target
@@ -39,7 +45,6 @@ def __init__(self, actor: str, verb, note_object: str, source: str, actor_name:
             a group name
         :param source: source service for the note. String.
         :param actor_name: Real name of the actor (or None)
-        :param actor_type: Type of actor. Allowed = 'user', 'group'
         :param level: The level of the notification. Allowed values are alert, warning, request,
             error (default "alert")
         :param target: target of the note. Optional. Should be a user id or group id if present.
@@ -60,21 +65,16 @@ def __init__(self, actor: str, verb, note_object: str, source: str, actor_name:
             * validate context fits
         """
         assert actor is not None, "actor must not be None"
-        assert actor_type in ['user', 'group'], "actor_type must be either 'user' or 'group'"
         assert verb is not None, "verb must not be None"
         assert note_object is not None, "note_object must not be None"
         assert source is not None, "source must not be None"
         assert level is not None, "level must not be None"
         assert target is None or isinstance(target, list), "target must be either a list or None"
         assert users is None or isinstance(users, list), "users must be either a list or None"
         assert context is None or isinstance(context, dict), "context must be either a dict or None"
-        assert actor_name is None or isinstance(actor_name, str), \
-            "actor_name must be either a str or None"
 
         self.id = str(uuid.uuid4())
         self.actor = actor
-        self.actor_name = actor_name
-        self.actor_type = actor_type
         self.verb = verbs.translate_verb(verb)
         self.object = note_object
         self.source = source
@@ -90,13 +90,16 @@ def __init__(self, actor: str, verb, note_object: str, source: str, actor_name:
         self.seen = seen
         self.users = users
 
-    def validate(self):
+    def validate(self) -> None:
         """
         Validates whether the notification fields are accurate. Should be called before
         sending a new notification to storage.
+        Raises exceptions if invalid. Currently just raises an InvalidExpirationError, as
+        that's all it checks...
         """
         self.validate_expiration(self.expires, self.created)
-        validate_actor(self.actor)
+        # TODO: do this later if we need to
+        # validate_actor(self.actor)
 
     def validate_expiration(self, expires: int, created: int):
         """
@@ -133,70 +136,80 @@ def to_dict(self) -> dict:
         """
         dict_form = {
             "id": self.id,
-            "actor": self.actor,
+            "actor": self.actor.to_dict(),
             "verb": self.verb.id,
-            "object": self.object,
+            "object": self.object.to_dict(),
             "source": self.source,
             "context": self.context,
-            "target": self.target,
             "level": self.level.id,
             "created": self.created,
             "expires": self.expires,
             "external_key": self.external_key,
-            "users": self.users,
-            "actor_type": self.actor_type
         }
+        target_dict = []
+        if self.target is not None:
+            target_dict = [t.to_dict() for t in self.target]
+            dict_form["target"] = target_dict
+        user_dict = []
+        if self.users is not None:
+            user_dict = [u.to_dict() for u in self.users]
+            dict_form["users"] = user_dict
+
         return dict_form
 
     def user_view(self) -> dict:
         """
         Returns a view of the Notification that's intended for the user.
-        That means we leave out the target and external keys.
+        That means we leave out the users and external keys.
         """
         view = {
             "id": self.id,
-            "actor": self.actor,
-            "actor_name": self.actor_name,
-            "actor_type": self.actor_type,
+            "actor": self.actor.to_dict(with_name=True),
             "verb": self.verb.past_tense,
-            "object": self.object,
+            "object": self.object.to_dict(with_name=True),
             "source": self.source,
             "context": self.context,
             "target": self.target,
             "level": self.level.name,
             "created": self.created,
             "expires": self.expires,
-            "seen": self.seen,
-            "external_key": self.external_key
+            "seen": self.seen
         }
+        target_dict = []
+        if self.target is not None:
+            target_dict = [t.to_dict(with_name=True) for t in self.target]
+            view["target"] = target_dict
+        user_dict = []
+        if self.users is not None:
+            user_dict = [u.to_dict(with_name=True) for u in self.users]
+            view["users"] = user_dict
+
         return view
 
     def serialize(self) -> str:
         """
-        Serializes this notification to a string for caching / simple storage.
+        Serializes this notification to a string for caching / simple storage (e.g. Redis).
         Assumes it's been validated.
         Just dumps it all to a json string.
         """
         serial = {
             "i": self.id,
-            "a": self.actor,
-            "an": self.actor_name,
-            "at": self.actor_type,
+            "a": str(self.actor),
             "v": self.verb.id,
-            "o": self.object,
+            "o": str(self.object),
             "s": self.source,
-            "t": self.target,
+            "t": [str(t) for t in self.target],
             "l": self.level.id,
             "c": self.created,
             "e": self.expires,
             "x": self.external_key,
             "n": self.context,
-            "u": self.users
+            "u": [str(u) for u in self.users]
         }
         return json.dumps(serial, separators=(',', ':'))
 
     @classmethod
-    def deserialize(cls, serial: str):
+    def deserialize(cls, serial: str) -> N:
         """
         Deserializes and returns a new Notification instance.
         """
@@ -212,26 +225,26 @@ def deserialize(cls, serial: str):
         missing_keys = required_keys.difference(struct.keys())
         if missing_keys:
             raise InvalidNotificationError('Missing keys: {}'.format(missing_keys))
+        users = [Entity.from_str(u) for u in struct.get('u', [])]
+        target = [Entity.from_str(t) for t in struct.get('t', [])]
         deserial = cls(
-            struct['a'],
+            Entity.from_str(struct['a']),
             str(struct['v']),
-            struct['o'],
+            Entity.from_str(struct['o']),
             struct['s'],
-            actor_type=struct.get('at', 'user'),
             level=str(struct['l']),
-            target=struct.get('t'),
+            target=target,
             context=struct.get('n'),
             external_key=struct.get('x'),
-            users=struct.get('u')
+            users=users
         )
         deserial.created = struct['c']
         deserial.id = struct['i']
         deserial.expires = struct['e']
-        deserial.actor_name = struct['an']
         return deserial
 
     @classmethod
-    def from_dict(cls, serial: dict):
+    def from_dict(cls, serial: dict) -> N:
         """
         Returns a new Notification from a serialized dictionary (e.g. used in Mongo)
         """
@@ -246,18 +259,16 @@ def from_dict(cls, serial: dict):
         if missing_keys:
             raise InvalidNotificationError('Missing keys: {}'.format(missing_keys))
         deserial = cls(
-            serial['actor'],
+            Entity.from_dict(serial['actor']),
             str(serial['verb']),
-            serial['object'],
+            Entity.from_dict(serial['object']),
             serial['source'],
             level=str(serial['level']),
-            target=serial.get('target'),
+            target=[Entity.from_dict(t) for t in serial.get('target', [])],
             context=serial.get('context'),
             external_key=serial.get('external_key'),
             seen=serial.get('seen', False),
-            users=serial.get('users'),
-            actor_name=serial.get('actor_name'),
-            actor_type=serial.get('actor_type', 'user')
+            users=[Entity.from_dict(u) for u in serial.get('users', [])]
         )
         deserial.created = serial['created']
         deserial.expires = serial['expires']

feeds/actor.py

@@ -6,6 +6,9 @@
     validate_user_id,
     validate_user_ids
 )
+from .external_api.groups import (
+    validate_group_id
+)
 from .exceptions import InvalidActorError
 
 
@@ -16,8 +19,10 @@ def validate_actor(actor, actor_type="user"):
         else:
             raise InvalidActorError("Actor '{}' is not a real user.".format(actor))
     elif actor_type == "group":
-        # pending api
-        return True
+        if validate_group_id(actor):
+            return True
+        else:
+            raise InvalidActorError("Actor '{}' is not a real group.".format(actor))
 
 
 def actor_ids_to_names(id_list: list):

feeds/api/admin_v1.py

@@ -18,6 +18,7 @@
     parse_notification_params,
     parse_expire_notifications_params
 )
+from feeds.entity import Entity
 
 cfg = get_config()
 admin_v1 = flask.Blueprint('admin_v1', __name__)
@@ -45,15 +46,15 @@ def add_global_notification():
 
     params = parse_notification_params(json.loads(request.get_data()), is_global=True)
     new_note = Notification(
-        'kbase',
+        Entity('kbase', 'admin'),
         params.get('verb'),
-        params.get('object'),
+        Entity('kbase', 'admin'),
         'kbase',
         level=params.get('level'),
         context=params.get('context'),
         expires=params.get('expires')
     )
-    global_feed = NotificationFeed(cfg.global_feed)
+    global_feed = NotificationFeed(cfg.global_feed, cfg.global_feed_type)
     global_feed.add_notification(new_note)
     return (flask.jsonify({'id': new_note.id}), 200)
 

feeds/api/api_v1.py

@@ -55,6 +55,7 @@ def get_notifications():
     1. validate/authenticate user
     2. make user feed object
     3. query user feed for most recent, based on params
+    #TODO: support group "feeds"
     """
     max_notes = request.args.get('n', default=10, type=int)
 
@@ -75,14 +76,14 @@ def get_notifications():
     include_seen = False if include_seen == 0 else True
     user_id = validate_user_token(get_auth_token(request))
     log(__name__, 'Getting feed for {}'.format(user_id))
-    feed = NotificationFeed(user_id)
+    feed = NotificationFeed(user_id, "user")
     user_notes = feed.get_notifications(
         count=max_notes, include_seen=include_seen, level=level_filter,
         verb=verb_filter, reverse=rev_sort, user_view=True
     )
 
     # fetch the globals
-    global_feed = NotificationFeed(cfg.global_feed)
+    global_feed = NotificationFeed(cfg.global_feed, cfg.global_feed_type)
     global_notes = global_feed.get_notifications(count=max_notes, user_view=True)
     return_vals = {
         "user": user_notes,
@@ -132,7 +133,6 @@ def add_notification():
         params.get('verb'),
         params.get('object'),
         params.get('source'),
-        actor_type=params.get('actor_type', 'user'),
         level=params.get('level'),
         target=params.get('target', []),
         context=params.get('context'),
@@ -150,7 +150,7 @@ def add_notification():
 @api_v1.route('/notifications/global', methods=['GET'])
 @cross_origin()
 def get_global_notifications():
-    global_feed = NotificationFeed(cfg.global_feed)
+    global_feed = NotificationFeed(cfg.global_feed, cfg.global_feed_type)
     global_notes = global_feed.get_notifications(user_view=True)
     return flask.jsonify(global_notes)
 
@@ -190,11 +190,11 @@ def get_single_notification(note_id):
     Should only return the note with that id if it's in the user's feed.
     """
     user_id = validate_user_token(get_auth_token(request))
-    feed = NotificationFeed(user_id)
+    feed = NotificationFeed(user_id, "user")
     try:
         note = feed.get_notification(note_id)
     except NotificationNotFoundError:
-        note = NotificationFeed(cfg.global_feed).get_notification(note_id)
+        note = NotificationFeed(cfg.global_feed, cfg.global_feed_type).get_notification(note_id)
     return (flask.jsonify({'notification': note.user_view()}), 200)
 
 
@@ -212,7 +212,7 @@ def mark_notifications_unseen():
     params = _get_mark_notification_params(json.loads(request.get_data()))
     note_ids = params.get('note_ids')
 
-    feed = NotificationFeed(user_id)
+    feed = NotificationFeed(user_id, "user")
     unauthorized_notes = list()
     for note_id in note_ids:
         try:
@@ -241,7 +241,7 @@ def mark_notifications_seen():
     params = _get_mark_notification_params(json.loads(request.get_data()))
     note_ids = params.get('note_ids')
 
-    feed = NotificationFeed(user_id)
+    feed = NotificationFeed(user_id, "user")
     unauthorized_notes = list()
     for note_id in note_ids:
         try:

feeds/api/util.py

@@ -2,6 +2,7 @@
     IllegalParameterError,
     MissingParameterError
 )
+from feeds.entity import Entity
 
 
 def parse_notification_params(params: dict, is_global: bool=False) -> dict:
@@ -10,27 +11,51 @@ def parse_notification_params(params: dict, is_global: bool=False) -> dict:
     Raises a MissingParameter error otherwise.
     Returns the params after parsing (currently does nothing, but if
     transformations are needed in the future, here's where that happens).
+    In total, can raise:
+    MissingParameterError (if required params are missing)
+    IllegalParameterError (if the wrong types are present)
+    EntityValidationError (if an Entity is malformed)
     """
-    # * `actor` - a user or org id.
-    # * `type` - one of the type keywords (see below, TBD (as of 10/8))
-    # * `target` - optional, a user or org id. - always receives this notification
-    # * `object` - object of the notice. For invitations, the group to be invited to.
-    #   For narratives, the narrative UPA.
+    # * `actor` - an Entity structure - gets turned into an Entity object when returned
+    # * `type` - one of the type keywords
+    # * `target` - optional, a list of Entity structures. This gets turned into a list of
+    #   entity object on return
+    # * `object` - object of the notice, an Entity structure. For invitations, the group to be
+    #   invited to. For narratives, the narrative UPA. Gets turned into an Entity object
+    #   when returned
+    # * `users` - a list of Entity objects, should be of either type user or group
     # * `level` - alert, error, warning, or request.
     # * `context` - optional, context of the notification, otherwise it'll be
     #   autogenerated from the info above.
 
     if not isinstance(params, dict):
         raise IllegalParameterError('Expected a JSON object as an input.')
-    required_list = ['verb', 'object', 'level']
+    required_list = ['verb', 'level']
     if not is_global:
-        required_list = required_list + ['actor', 'target', 'source']
+        required_list = required_list + ['actor', 'source', 'object']
     missing = [r for r in required_list if r not in params or params.get(r) is None]
     if missing:
         raise MissingParameterError("Missing parameter{} - {}".format(
             "s" if len(missing) > 1 else '',
             ", ".join(missing)
         ))
+    if not is_global:
+        # do the entity transformations
+        # If there are any EntityValidationErrors, they'll pop on up.
+        params["actor"] = Entity.from_dict(params["actor"])
+        params["object"] = Entity.from_dict(params["object"])
+        if "target" in params:
+            target = params["target"]
+            if isinstance(target, list):
+                params["target"] = [Entity.from_dict(t) for t in target]
+            else:
+                raise IllegalParameterError("Expected target to be a list of Entity structures.")
+        if "users" in params:
+            users = params["users"]
+            if isinstance(users, list):
+                params["users"] = [Entity.from_dict(u) for u in users]
+            else:
+                raise IllegalParameterError("Expected users to be a list of Entity structures.")
     return params
 
 

feeds/config.py

@@ -53,6 +53,7 @@ def __init__(self):
         self.db_pw = self._get_line(cfg, KEY_DB_PW, required=False)
         self.db_name = self._get_line(cfg, KEY_DB_NAME, required=False)
         self.global_feed = self._get_line(cfg, KEY_GLOBAL_FEED)
+        self.global_feed_type = "user"  # doesn't matter, need a valid Entity type...
         self.lifespan = self._get_line(cfg, KEY_LIFESPAN)
         try:
             self.lifespan = int(self._get_line(cfg, KEY_LIFESPAN))

feeds/exceptions.py

@@ -105,3 +105,38 @@ class NotificationNotFoundError(Exception):
     or isn't in the user's feed.
     """
     pass
+
+
+class GroupsError(Exception):
+    """
+    Generic exception wrapped around any response from the Groups server.
+    """
+    pass
+
+
+class EntityNameError(Exception):
+    """
+    Exception thrown when a failure to find an entity name happens
+    """
+    pass
+
+
+class EntityValidationError(Exception):
+    """
+    Thrown when an entity is unable to be validated
+    """
+    pass
+
+
+class WorkspaceError(Exception):
+    """
+    Generic exception wrapper for Workspace calls.
+    """
+    pass
+
+
+class JobError(Exception):
+    """
+    Generic wrapper around exceptions from Job Service calls.
+    """
+    pass

feeds/external_api/groups.py

@@ -1,10 +1,11 @@
 from ..config import get_config
-# import json
 import requests
 from typing import (
     List,
     Dict
 )
+from feeds.exceptions import GroupsError
+
 config = get_config()
 GROUPS_URL = config.groups_url
 
@@ -31,7 +32,13 @@ def validate_group_id(group_id: str) -> bool:
     """
     r = __groups_request("/group/{}/exists".format(group_id))
     res = r.json()
-    return res.get('exists', False)
+    if 'exists' in res:
+        return res['exists']
+    elif 'error' in res:
+        raise GroupsError(
+            "Error while looking up group id: " +
+            res['error'].get('message', 'no message available')
+        )
 
 
 def __groups_request(path: str, token: str=None) -> Response:

feeds/feeds/notification/notification_feed.py

@@ -5,13 +5,14 @@
 from cachetools import TTLCache
 import logging
 from feeds.exceptions import NotificationNotFoundError
-from feeds.actor import actor_ids_to_names
+# from feeds.actor import actor_ids_to_names
+from feeds.entity import Entity
 
 
 class NotificationFeed(BaseFeed):
-    def __init__(self, user_id):
-        self.user_id = user_id
-        self.timeline_storage = MongoTimelineStorage(self.user_id)
+    def __init__(self, user_id, user_type):
+        self.user = Entity(user_id, user_type)
+        self.timeline_storage = MongoTimelineStorage(user_id, user_type)
         self.activity_storage = MongoActivityStorage()
         self.timeline = None
         self.cache = TTLCache(1000, 600)
@@ -24,7 +25,9 @@ def _update_timeline(self):
 
         TODO: add metadata to timeline storage - type and verb, first.
         """
-        logging.getLogger(__name__).info('Fetching timeline for ' + self.user_id)
+        logging.getLogger(__name__).info(
+            'Fetching timeline for '.format(self.user)
+        )
         self.timeline = self.timeline_storage.get_timeline()
 
     def get_notifications(self, count: int=10, include_seen: bool=False, level=None, verb=None,
@@ -92,17 +95,17 @@ def get_activities(self, count=10, include_seen=False, level=None, verb=None,
             level=level, verb=verb, reverse=reverse
         )
         note_list = list()
-        actor_ids = set()
+        # actor_ids = set()
         for note in serial_notes:
-            actor_ids.add(note["actor"])
-            if self.user_id not in note["unseen"]:
-                note["seen"] = True
-            else:
-                note["seen"] = False
+            # actor_ids.add(note["actor"].id)
+            # if self.user_id not in note["unseen"]:
+            #     note["seen"] = True
+            # else:
+            #     note["seen"] = False
             note_list.append(Notification.from_dict(note))
-        actor_names = actor_ids_to_names(list(actor_ids))
-        for note in note_list:
-            note.actor_name = actor_names.get(note.actor, {}).get("name")
+        # actor_names = actor_ids_to_names(list(actor_ids))
+        # for note in note_list:
+        #     note.actor_name = actor_names.get(note.actor, {}).get("name")
         return note_list
 
     def mark_activities(self, activity_ids, seen=False):
@@ -112,9 +115,9 @@ def mark_activities(self, activity_ids, seen=False):
         changed for that activity.
         """
         if seen:
-            self.activity_storage.set_seen(activity_ids, self.user_id)
+            self.activity_storage.set_seen(activity_ids, self.user)
         else:
-            self.activity_storage.set_unseen(activity_ids, self.user_id)
+            self.activity_storage.set_unseen(activity_ids, self.user)
 
     def add_notification(self, note):
         return self.add_activity(note)
@@ -123,7 +126,7 @@ def add_activity(self, note):
         """
         Adds an activity to this user's feed
         """
-        self.activity_storage.add_to_storage(note, [self.user_id])
+        self.activity_storage.add_to_storage(note, [self.user])
 
     def get_unseen_count(self):
         """

feeds/managers/fanout_modules/base.py

@@ -1,5 +1,7 @@
 from abc import abstractmethod
 from feeds.activity.notification import Notification
+from feeds.entity import Entity
+from typing import List
 
 
 class FanoutModule(object):
@@ -12,9 +14,9 @@ def __init__(self, note: Notification):
         self.note = note
 
     @abstractmethod
-    def get_target_users(self):
+    def get_target_users(self) -> List[Entity]:
         """
-        This should always return a list, even an empty one.
+        This should always return a list of Entities, even an empty one.
         Ideally, it'll be a list of users that should see the notification.
         """
         pass

feeds/managers/fanout_modules/kbase.py

@@ -1,7 +1,9 @@
 from .base import FanoutModule
 from feeds.config import get_config
+from feeds.entity import Entity
 
 
 class KBaseFanout(FanoutModule):
     def get_target_users(self):
-        return [get_config().global_feed]
+        cfg = get_config()
+        return [Entity(cfg.global_feed, cfg.global_feed_type)]

feeds/managers/notification_manager.py

@@ -17,6 +17,7 @@
 from .fanout_modules.workspace import WorkspaceFanout
 from .fanout_modules.jobs import JobsFanout
 from .fanout_modules.kbase import KBaseFanout
+from feeds.entity import Entity
 
 
 class NotificationManager(BaseManager):
@@ -35,7 +36,7 @@ def add_notification(self, note: Notification):
         activity_storage = MongoActivityStorage()
         activity_storage.add_to_storage(note, target_users)
 
-    def get_target_users(self, note: Notification) -> List[str]:
+    def get_target_users(self, note: Notification) -> List[Entity]:
         """
         This is gonna get complex.
         The target users are a combination of:
@@ -62,7 +63,7 @@ def get_target_users(self, note: Notification) -> List[str]:
         return user_list
 
     def expire_notifications(self, note_ids: list, external_keys: list, source: str=None,
-                             is_admin: bool=False):
+                             is_admin: bool=False) -> Dict[str, list]:
         """
         Expires notifications.
         All notifications identified by either their id, or external key, must come from the same
@@ -104,7 +105,8 @@ def expire_notifications(self, note_ids: list, external_keys: list, source: str=
             "expired": expired
         }
 
-    def get_notifications_by_ext_keys(self, external_keys: List[str], source: str) -> Dict:
+    def get_notifications_by_ext_keys(self, external_keys: List[str],
+                                      source: str) -> Dict[str, Notification]:
         """
         Fetches notifications by their external key and source.
         These are returned as a dictionary where the keys are the

feeds/storage/base.py

@@ -24,9 +24,11 @@ def remove_from_storage(self, activity_ids):
 
 
 class TimelineStorage(BaseStorage):
-    def __init__(self, user_id):
+    def __init__(self, user_id, user_type):
         assert user_id
         self.user_id = user_id
+        assert user_type
+        self.user_type = user_type  # should align with entity types
 
     def add_to_timeline(self, activity):
         raise NotImplementedError()

feeds/storage/mongodb/activity_storage.py

@@ -1,44 +1,48 @@
-from typing import List
+from typing import (
+    List,
+    Dict
+)
 from ..base import ActivityStorage
 from .connection import get_feeds_collection
 from feeds.exceptions import (
     ActivityStorageError
 )
 from pymongo.errors import PyMongoError
 from feeds.util import epoch_ms
+from feeds.entity import Entity
 
 
 class MongoActivityStorage(ActivityStorage):
-    def add_to_storage(self, activity, target_users: List[str]):
+    def add_to_storage(self, activity, target_users: List[Entity]) -> None:
         """
         Adds a single activity to the MongoDB.
         Returns None if successful.
         Raises an ActivityStorageError if it fails.
         """
         coll = get_feeds_collection()
         act_doc = activity.to_dict()
-        act_doc["users"] = target_users
-        act_doc["unseen"] = target_users
+        act_doc["users"] = [t.to_dict() for t in target_users]
+        act_doc["unseen"] = [t.to_dict() for t in target_users]
         try:
             coll.insert_one(act_doc)
         except PyMongoError as e:
             raise ActivityStorageError("Failed to store activity: " + str(e))
 
-    def set_unseen(self, act_ids: List[str], user: str):
+    def set_unseen(self, act_ids: List[str], user: Entity) -> None:
         """
         Setting unseen means adding the user to the list of unseens. But we should only do that for
         docs that the user can't see anyway, so put that in the query.
         """
         coll = get_feeds_collection()
         coll.update_many({
             'id': {'$in': act_ids},
-            'users': user,
+            'users': user.to_dict(),
             'unseen': {'$nin': [user]}
         }, {
             '$addToSet': {'unseen': user}
         })
 
-    def set_seen(self, act_ids: List[str], user: str):
+    def set_seen(self, act_ids: List[str], user: Entity) -> None:
         """
         Setting seen just means removing the user from the list of unseens.
         The query should find all docs in the list of act_ids, where the user
@@ -48,13 +52,13 @@ def set_seen(self, act_ids: List[str], user: str):
         coll = get_feeds_collection()
         coll.update_many({
             'id': {'$in': act_ids},
-            'users': user,
+            'users': user.to_dict(),
             'unseen': {'$all': [user]}
         }, {
             '$pull': {'unseen': user}
         })
 
-    def get_by_id(self, act_ids: List[str], source: str=None):
+    def get_by_id(self, act_ids: List[str], source: str=None) -> Dict[str, dict]:
         """
         If source is not None, return only those that match the source.
         Returns a dict mapping from note id to note
@@ -73,7 +77,7 @@ def get_by_id(self, act_ids: List[str], source: str=None):
             notes[d["id"]] = d
         return notes
 
-    def get_by_external_key(self, external_keys: List[str], source):
+    def get_by_external_key(self, external_keys: List[str], source: str) -> Dict[str, dict]:
         """
         Source HAS to exist here, it's part of the index.
         Returns a dict mapping from external_key to note
@@ -92,7 +96,7 @@ def get_by_external_key(self, external_keys: List[str], source):
             notes[d["external_key"]] = d
         return notes
 
-    def expire_notifications(self, act_ids: List[str]):
+    def expire_notifications(self, act_ids: List[str]) -> None:
         """
         Expires notifications by changing their expiration time to now.
         """

feeds/storage/mongodb/timeline_storage.py

@@ -2,33 +2,39 @@
 from ..base import TimelineStorage
 from .connection import get_feeds_collection
 from feeds.util import epoch_ms
+from feeds.activity.base import BaseActivity
+from feeds.notification_level import Level
+from feeds.verbs import Verb
+from typing import (
+    List,
+    Dict
+)
 
 
 class MongoTimelineStorage(TimelineStorage):
     """
     Returns the serialized/dictified storage elements, not the actual Activity objects.
     Meant to be able to deal with Activities that aren't just Notifications.
     """
-    def add_to_timeline(self, activity):
+    def add_to_timeline(self, activity: BaseActivity) -> None:
         raise NotImplementedError()
 
-    def get_timeline(self, count: int=10, include_seen: int=False, level=None, verb=None,
-                     reverse: bool=False) -> list:
+    def get_timeline(self, count: int=10, include_seen: int=False, level: Level=None,
+                     verb: Verb=None, reverse: bool=False) -> List[dict]:
         """
         :param count: int > 0
         :param include_seen: boolean
         :param level: Level or None
         :param verb: Verb or None
         """
-        # TODO: input validation
         coll = get_feeds_collection()
         now = epoch_ms()
         query = {
-            "users": self.user_id,
+            "users": self._user_doc(),
             "expires": {"$gt": now}
         }
         if not include_seen:
-            query['unseen'] = self.user_id
+            query['unseen'] = self._user_doc()
         if level is not None:
             query['level'] = level.id
         if verb is not None:
@@ -40,11 +46,11 @@ def get_timeline(self, count: int=10, include_seen: int=False, level=None, verb=
         serial_notes = [note for note in timeline]
         return serial_notes
 
-    def get_single_activity_from_timeline(self, note_id):
+    def get_single_activity_from_timeline(self, note_id: str) -> dict:
         coll = get_feeds_collection()
         query = {
             "id": note_id,
-            "users": self.user_id
+            "users": self._user_doc()
         }
         note_serial = coll.find_one(query)
         if note_serial is None:
@@ -55,12 +61,18 @@ def get_single_activity_from_timeline(self, note_id):
             note_serial['seen'] = True
         return note_serial
 
-    def get_unseen_count(self):
+    def get_unseen_count(self) -> int:
         coll = get_feeds_collection()
         now = epoch_ms()
         query = {
-            "users": self.user_id,
-            "unseen": self.user_id,
+            "users": self._user_doc(),
+            "unseen": self._user_doc(),
             "expires": {"$gt": now}
         }
         return coll.find(query).count()
+
+    def _user_doc(self) -> Dict[str, str]:
+        return {
+            "id": self.user_id,
+            "type": self.user_type
+        }

test/_data/mongo/notifications.json

@@ -233,13 +233,25 @@ def test_expire_notification_admin_from_service(client, mongo_notes, mock_valid_
     # service creates two notifications - one with external key
     service_cred = {"Authorization": "token-"+str(uuid4())}
     note = {
-        "actor": "kbasetest",
+        "actor": {
+            "id": "kbasetest",
+            "type": "user"
+        },
         "verb": 1,
         "level": 1,
-        "object": "stuff",
-        "users": ["kbasetest"],
+        "object": {
+            "id": "stuff",
+            "type": "workspace"
+        },
+        "users": [{
+            "id": "kbasetest",
+            "type": "user"
+        }],
         "source": source,
-        "target": ["kbasetest"]
+        "target": [{
+            "id": "kbasetest",
+            "type": "user"
+        }]
     }
     response = client.post(
         "/api/V1/notification",

test/api/test_admin_v1.py

@@ -2,12 +2,14 @@
 from feeds.feeds.notification.notification_feed import NotificationFeed
 from feeds.activity.notification import Notification
 
+USER = "test_user"
+USER_TYPE = "user"
+
 def test_get_notifications(mongo_notes):
     """
     Imported the dataset. The main user is test_user, so get their feed in various ways.
     """
-    user = "test_user"
-    feed = NotificationFeed(user)
+    feed = NotificationFeed(USER, USER_TYPE)
     # as NOT user_view, should be a list of Notification objects
     notes = feed.get_notifications()
     assert "feed" in notes and len(notes["feed"]) == 7
@@ -16,15 +18,13 @@ def test_get_notifications(mongo_notes):
         assert isinstance(n, Notification)
 
 def test_get_notifications_fail(mongo_notes):
-    user = "test_user"
-    feed = NotificationFeed(user)
+    feed = NotificationFeed(USER, USER_TYPE)
     with pytest.raises(ValueError) as e:
         feed.get_notifications(count=0)
     assert "Count must be an integer > 0" == str(e.value)
 
 def test_update_timeline(mongo_notes):
-    user = "test_user"
-    feed = NotificationFeed(user)
+    feed = NotificationFeed(USER, USER_TYPE)
     assert feed.timeline is None
     feed._update_timeline()
     assert feed.timeline is not None

test/feeds/notification/test_notification_feed.py

@@ -178,7 +178,7 @@ def test_server_missing_params(client, mock_valid_service_token):
     _validate_error(data, {
         'http_code': 422,
         'http_status': 'Unprocessable Entity',
-        'message': 'Missing parameters - verb, object, level, target, source'
+        'message': 'Missing parameters - verb, level, source, object'
     })