Stream Framework is a Python library, which allows you to build news feed, activity streams and notification systems using Cassandra and/or Redis. The authors of Stream-Framework also provide a cloud service for feed technology:

I have created a CustomAggregatedActivity extended from feedly.activity.AggregatedActivity:

from feedly.activity import AggregatedActivity

class CustomAggregatedActivity(AggregatedActivity):
    @property
    def abc(self):
        pass

And then i have used this in my custom aggregator:

class CustomAggregator(BaseAggregator):
    aggregation_class = CustomAggregatedActivity

    def rank(self, aggregated_activities):
        # rank logic here

     def group(self, activity):
         # group logic here

Then i have assigned the CustomAggregator in:

class AggregatedUserFeed(RedisAggregatedFeed):
    aggregator_class = CustomAggregator
    key_format = 'feed:aggregated:%(user_id)s'

and finally:

class Newsfeed(Feedly):
    feed_classes = dict(
        normal=NormalUserFeed,
        aggregated=AggregatedUserFeed
    )
    user_feed_class = UserFeed
newsfeed_stream = Newsfeed()

Now when i get user aggregated feed:

feed = newsfeed_stream.get_feeds(user_id)['aggregated']
aggregated_activities = list(feed[:10])
>>> aggregated_activities[0].abc

It says:

AttributeError: 'AggregatedActivity' object has no attribute 'abc'

Can you tell me why this happened?

Thanks!

Hi again! Im working on a django project (more info in issue #16 ) I've got my Celery worker up and running, my redis seems database ok and most of my setup seems fine.. The only problem is that when I try to "fanout" to new feeds, the feeds does not seem to reach the Celery worker.. (seems no tasks are added to the worker queue). I know this might be a broker/worker issue, but I haven't found any good indications of what the problem might be :( Everything works fine when CELERY_ALWAYS_EAGER == True , so I don't think there is a problem with my code..

for feedly to work properly with celery and redis the things I would need is a Celery worker running, redis db setup ok, and a task broker (redis or rabbitmq) , right? I think I lack the understanding of how the broker and worker works together with feedly..?

Any tips are much appreciated :)

Adding to feed: activity = Activity(wishlist.user,WishVerb,in_prod.id) feed = UserWishFeed(wishlist.user.id) feed.insert_activity(activity) feed.add(activity) feedly.add_wish(wishlist.user, activity)

A view for fetching friends-feed : @csrf_exempt @api_view(['GET']) @login_required def friends_wish_feed(request, _args, *_kwargs): user = GiftItEndUser.objects.get(pk=kwargs['pk']) if request.user.id != user.pk: content = {"message":"you do not have permission","status":0} json_data = json.dumps(content) return Response(json_data)

    feed = feedly.get_feeds(request.user.id)['normal']
    act_list = []
    json_data = []
    activities = list(feed[:25])
    for act in activities:
            act = {'user_id':act.actor_id,'product_id':act.object_id}
            act_list.append(act)
            json_data = json.dumps(act_list)
    return Response(json_data)

Some celery settings: FEEDLY_NYDUS_CONFIG = { 'CONNECTIONS': { 'redis': { 'engine': 'nydus.db.backends.redis.Redis', 'router': 'nydus.db.routers.redis.PrefixPartitionRouter', 'hosts': { 0: {'prefix': 'default', 'db': 0, 'host': '127.0.0.1', 'port': 6379}, } }, } }

BROKER_URL = 'redis://localhost:6379/0' CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

Further to the conversation in #174, I saw that there's now an option to install without cassandra pip install stream-framework --install-option="--no-cassandra", however I don't think this option makes it possible to install via a requirements file without cassandra.

This pull request makes it possible to install via

$ pip install Stream-Framework[cassandra2]

$ pip install Stream-Framework[cassandra3]

$ pip install Stream-Framework[redis]

And these options can also be used in a requirements file, eg

Stream-Framework[cassandra2]==1.3.4

Hi team,based on this instruction From the root of the feedly project run:

vagrant up vagrant provision vagrant ssh cd pinterest_example python manage.py runserver 0:8000

I can't see to find the pinterest_example folder, why?

Just for testing i have set feed max_length=1000 and merge_max_length=1000, why? because duplicate aggregated activities is not an option for me in newsfeed (see on facebook newsfeed you will never see a duplicate activity). So i ran a program which reads existing user actions from database and generates feeds using celery (async tasks).

I was expecting that max_length=merge_max_length would never create a duplicate aggregated activity, but the program issue so many celery tasks that in the end there were some duplicate aggregated activities. Why that happen?

Why there is even merge_max_length variable? I thought we were storing activities in redis as key, value where key is a group? so why there is a need to traverse through all aggregated activities to find where the new aggregated activity belong? A new aggregated activity already have group defined so why not do something like this as we do in python dict.get(group)?

Hello there,

I'm trying to setup count publishing via pubsub on our project. It seems to work, but I've got a question:

https://github.com/tschellenbach/Stream-Framework/blob/master/stream_framework/feeds/aggregated_feed/notification_feed.py#L80

Why is there global channel used? Are there some reasons like performance or something? It looks like there should be just something like:

 self.redis.publish(self.pubsub_key, count_data)

Also, it seems like there is remove_many() override missing with self.denormalize_count() call in it (same as in add_many())

Thank you guys for a great lib!

I have an issue similar to that in issue #8 .

Im trying to make a REST service with this, I use django-rest-framework.

Seems adding activities to redis works, but when i try to fetch them back i will get this

KeyError: 13869294460000000000005005L

seems like the "L" is causing some sort of error ?

function calls and stack trace: wishlists/views.py

feedly = UserWishFeedly()

//function for adding activities: activity = Activity(wishlist.user, WishVerb, in_prod.id) feed = UserWishFeed(wishlist.user.id) feed.add(activity)

//test-function for getting the feed def wish_feed(request,args,*kwargs): feed = feedly.get_user_feed(request.user.id) activities = feed[:10] content = {"activities":activities} json_data = json.dumps(content) return Response(json_data)

File "/usr/local/lib/python2.7/dist-packages/django/core/handlers/base.py", line 115, in get_response response = callback(request, _callback_args, *_callback_kwargs)

File "/usr/local/lib/python2.7/dist-packages/django/views/decorators/csrf.py", line 77, in wrapped_view return view_func(_args, *_kwargs)

File "/usr/local/lib/python2.7/dist-packages/django/views/generic/base.py", line 68, in view return self.dispatch(request, _args, *_kwargs)

File "/usr/local/lib/python2.7/dist-packages/django/views/decorators/csrf.py", line 77, in wrapped_view return view_func(_args, *_kwargs)

File "/usr/local/lib/python2.7/dist-packages/rest_framework/views.py", line 327, in dispatch response = self.handle_exception(exc)

File "/usr/local/lib/python2.7/dist-packages/rest_framework/views.py", line 324, in dispatch response = handler(request, _args, *_kwargs)

File "/usr/local/lib/python2.7/dist-packages/rest_framework/decorators.py", line 49, in handler return func(_args, *_kwargs)

File "/usr/local/lib/python2.7/dist-packages/django/contrib/auth/decorators.py", line 25, in _wrapped_view return view_func(request, _args, *_kwargs)

File "/home/andrenro/dev_giftit_webapp/env/Scripts/giftit_webapp/wishlists/views.py", line 437, in wish_feed activities = feed[:10]

File "/usr/local/lib/python2.7/dist-packages/feedly/feeds/base.py", line 261, in get item start, bound)

File "/usr/local/lib/python2.7/dist-packages/feedly/feeds/base.py", line 304, in get_activity_slice activities = self.hydrate_activities(activities)

File "/usr/local/lib/python2.7/dist-packages/feedly/feeds/base.py", line 285, in hydrate_activities return [activity.get_hydrated(activity_data) for activity in activities]

File "/usr/local/lib/python2.7/dist-packages/feedly/activity.py", line 41, in get_hydrated activity = activities[int(self.serialization_id)]

KeyError: 13869294460000000000005005L

If i try redis-cli i get:

127.0.0.1:6379> zrange "feed:user:6" 0 25

1. "13869294460000000000005005"

Im running python 2.7.3 and django 1.5.4 Im kinda stuck with this, hope someone can shed some light on it. Great project btw!!

It's not very clear how to implement getstream.io dynamic feed slug feature. Create dynamic feed class? or merge feed_slug into user_id like Feed(user_id="%(feed_slug)s:%(user_id)s") ? this is too hack.

Hi, first of all a little of what I am trying to achieve. I have built a django wrapper on top of stream-framework library. There are 2 feed-classes - FlatFeed(RedisFeed) and AggregatedFeed(RedisAggregatedFeed). Obviously, these are using Redis to store the feed data. I have also implemented my own aggregator class.

Error: The generated aggregated feed doesn't contain all activities while the flat feed has all the activities. The use-case is - there are 3 users A, B and C. B and C performs some activities, then user A follows B and C. User B and C keep on doing more activities. Flat feed of A contains all the activities of B and C, but aggregated feed of A has some lost activities.

For example, B likes products 1, 2, 3, 4 C likes products 5, 6, 7, 8, 9, 10 A follows B A follows C

flat_feed(A) has all activities, but aggregated_feed(A) only has likes for 1, 5 and 8. I repeated this use-case several times and each time only these 3 activities are coming.

I have tested my aggregator class implementation on django shell. The output of aggregate and merge function contains all the activities.

Please help !!

Please note that flat feed has correct entries, the missing entries are only in aggregated feeds.

For reusable apps shared_task are used. Also for me with celery.task.task decorator, celery was using default settings instead of setting from project.

http://celery.readthedocs.org/en/latest/django/first-steps-with-django.html#using-the-shared-task-decorator

Hi,

It seems like serialization ids mismatch by 'L' letter, so what would be the possible cause? Is feedly hitting redis twice? Here is the step in trace of the run

> get_user_feed(customer.user.id)[:10]
[Activity(joined) 7 5]
(Pdb) activity_list = self.activity_storage.get_many(activity_ids)
(Pdb) activity_ids
['13778137560000000000005005']
(Pdb) activity_data
{13778173560000000000005005L: Activity(joined) 7 5}
-> return [activity.get_hydrated(activity_data) for activity in activities]
KeyError: (13778137560000000000005005L,)

Feedly version: feedly==0.9.3

Thanks

There are small typos in:

• docs/conf.py
• docs/readme.rst
• stream_framework/tests/storage/base.py

Fixes:

• Should read that rather than shat.
• Should read pollution rather than polution.
• Should read immediately rather than immediatly.

Semi-automated pull request generated by https://github.com/timgates42/meticulous/blob/master/docs/NOTE.md

ImportError: cannot import name 'BasePipeline' from 'redis.client' (/home/ian/.local/lib/python3.8/site-packages/redis/client.py)

ian@ian-HP-Stream-Laptop-11-y0XX:~/stream-framework/Stream-Framework$ python3 Python 3.8.10 (default, Sep 28 2021, 16:10:42) [GCC 9.3.0] on linux Type "help", "copyright", "credits" or "license" for more information.

from stream_framework.activity import Activity

def create_activity(pin): ... activity = Activity( ... pin.user_id, ... PinVerb, ... pin.id, ... pin.influencer_id, ... time=make_naive(pin.created_at, pytz.utc), ... extra_context=dict(item_id=pin.item_id) ... ) ... return activity ... from stream_framework.feeds.redis import RedisFeed Traceback (most recent call last): File "", line 1, in File "/home/ian/stream-framework/Stream-Framework/stream_framework/feeds/redis.py", line 2, in from stream_framework.storage.redis.activity_storage import RedisActivityStorage File "/home/ian/stream-framework/Stream-Framework/stream_framework/storage/redis/activity_storage.py", line 2, in from stream_framework.storage.redis.structures.hash import ShardedHashCache File "/home/ian/stream-framework/Stream-Framework/stream_framework/storage/redis/structures/hash.py", line 1, in from stream_framework.storage.redis.structures.base import RedisCache File "/home/ian/stream-framework/Stream-Framework/stream_framework/storage/redis/structures/base.py", line 2, in from redis.client import BasePipeline ImportError: cannot import name 'BasePipeline' from 'redis.client' (/home/ian/.local/lib/python3.8/site-packages/redis/client.py)

It's clear that I can use stream-framework for Notification/Activity feeds. However, the use-case I'm building involves a slightly different scenario

1. A per-user ranked feed of target objects. An example of target objects could be movie listings.
2. A per-user ranked, filtered feed of users who follow a target object

As far as 2) is concerned, it seems like I can override get_user_follower_ids to accomplish this [user->target-objects-the-user-is-following -> user_ids that follow the target objects]

Would this be a good way to handle this scenario ?