From 20ffa41a3a069d8b4acf0eaf325408dab3f9612d Mon Sep 17 00:00:00 2001 From: Roman Zavadski Date: Fri, 12 Nov 2021 21:09:05 +0300 Subject: [PATCH 1/3] added async support --- CHANGELOG.md | 5 + setup.py | 5 +- stream/__init__.py | 17 +- stream/client/__init__.py | 2 + stream/client/async_client.py | 276 ++++ stream/{client.py => client/base.py} | 601 ++++----- stream/client/client.py | 289 +++++ stream/collections.py | 128 -- stream/collections/__init__.py | 1 + stream/collections/base.py | 100 ++ stream/collections/collections.py | 150 +++ stream/exceptions.py | 4 +- stream/feed.py | 233 ---- stream/feed/__init__.py | 1 + stream/feed/base.py | 172 +++ stream/feed/feeds.py | 238 ++++ stream/personalization.py | 66 - stream/personalization/__init__.py | 1 + stream/personalization/base.py | 30 + stream/personalization/personalizations.py | 117 ++ stream/reactions.py | 89 -- stream/reactions/__init__.py | 1 + stream/reactions/base.py | 75 ++ stream/reactions/reaction.py | 163 +++ stream/tests/conftest.py | 60 + stream/tests/test_async_client.py | 1342 ++++++++++++++++++++ stream/tests/test_client.py | 10 +- stream/users.py | 36 - stream/users/__init__.py | 1 + stream/users/base.py | 39 + stream/users/user.py | 73 ++ stream/utils.py | 15 +- 32 files changed, 3416 insertions(+), 924 deletions(-) create mode 100644 stream/client/__init__.py create mode 100644 stream/client/async_client.py rename stream/{client.py => client/base.py} (56%) create mode 100644 stream/client/client.py delete mode 100644 stream/collections.py create mode 100644 stream/collections/__init__.py create mode 100644 stream/collections/base.py create mode 100644 stream/collections/collections.py delete mode 100644 stream/feed.py create mode 100644 stream/feed/__init__.py create mode 100644 stream/feed/base.py create mode 100644 stream/feed/feeds.py delete mode 100644 stream/personalization.py create mode 100644 stream/personalization/__init__.py create mode 100644 stream/personalization/base.py create mode 100644 stream/personalization/personalizations.py delete mode 100644 stream/reactions.py create mode 100644 stream/reactions/__init__.py create mode 100644 stream/reactions/base.py create mode 100644 stream/reactions/reaction.py create mode 100644 stream/tests/conftest.py create mode 100644 stream/tests/test_async_client.py delete mode 100644 stream/users.py create mode 100644 stream/users/__init__.py create mode 100644 stream/users/base.py create mode 100644 stream/users/user.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 97e250e..33e22be 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,8 @@ +## 6.0.0 - 2021-01-19 + +* Add asynchronous code support +* Change ``%s`` to f-strings + ## 5.1.1 - 2022-01-18 * Handle backward compatible pyjwt 1.x support for token generation diff --git a/setup.py b/setup.py index a0835e3..f674f65 100644 --- a/setup.py +++ b/setup.py @@ -6,7 +6,9 @@ from stream import __version__, __maintainer__, __email__, __license__ import sys -tests_require = ["pytest", "unittest2", "pytest-cov", "python-dateutil"] +tests_require = [ + "pytest", "unittest2", "pytest-cov", "python-dateutil", "pytest-asyncio", +] ci_require = ["black", "flake8", "pytest-cov"] long_description = open("README.md", "r").read() @@ -16,6 +18,7 @@ "requests>=2.3.0,<3", "pyjwt>=2.0.0,<3", "pytz>=2019.3", + "aiohttp>=3.8.0", ] diff --git a/stream/__init__.py b/stream/__init__.py index 87ad6f1..6261e94 100644 --- a/stream/__init__.py +++ b/stream/__init__.py @@ -5,7 +5,7 @@ __copyright__ = "Copyright 2014, Stream.io, Inc" __credits__ = ["Thierry Schellenbach, mellowmorning.com, @tschellenbach"] __license__ = "BSD-3-Clause" -__version__ = "5.1.1" +__version__ = "6.0.0" __maintainer__ = "Thierry Schellenbach" __email__ = "support@getstream.io" __status__ = "Production" @@ -19,6 +19,7 @@ def connect( timeout=3.0, location=None, base_url=None, + use_async=False, ): """ Returns a Client object @@ -26,8 +27,9 @@ def connect( :param api_key: your api key or heroku url :param api_secret: the api secret :param app_id: the app id (used for listening to feed changes) + :param use_async: flag to set AsyncClient """ - from stream.client import StreamClient + from stream.client import AsyncStreamClient, StreamClient stream_url = os.environ.get("STREAM_URL") # support for the heroku STREAM_URL syntax @@ -42,6 +44,17 @@ def connect( else: raise ValueError("Invalid api key or heroku url") + if use_async: + return AsyncStreamClient( + api_key, + api_secret, + app_id, + version, + timeout, + location=location, + base_url=base_url, + ) + return StreamClient( api_key, api_secret, diff --git a/stream/client/__init__.py b/stream/client/__init__.py new file mode 100644 index 0000000..5d8511e --- /dev/null +++ b/stream/client/__init__.py @@ -0,0 +1,2 @@ +from .async_client import AsyncStreamClient +from .client import StreamClient diff --git a/stream/client/async_client.py b/stream/client/async_client.py new file mode 100644 index 0000000..7fa02c5 --- /dev/null +++ b/stream/client/async_client.py @@ -0,0 +1,276 @@ +import logging + +import aiohttp +from aiohttp import ClientConnectionError + +from stream import serializer +from stream.client.base import BaseStreamClient +from stream.collections import AsyncCollections +from stream.feed.feeds import AsyncFeed +from stream.personalization import AsyncPersonalization +from stream.reactions import AsyncReactions +from stream.serializer import _datetime_encoder +from stream.users import AsyncUsers +from stream.utils import ( + get_reaction_params, + validate_feed_slug, + validate_foreign_id_time, + validate_user_id, +) + +logger = logging.getLogger(__name__) + + +class AsyncStreamClient(BaseStreamClient): + def __init__( + self, + api_key, + api_secret, + app_id, + version="v1.0", + timeout=6.0, + base_url=None, + location=None, + ): + super().__init__( + api_key, + api_secret, + app_id, + version=version, + timeout=timeout, + base_url=base_url, + location=location, + ) + self.session = aiohttp.ClientSession + + token = self.create_jwt_token("collections", "*", feed_id="*", user_id="*") + self.collections = AsyncCollections(self, token) + + token = self.create_jwt_token("personalization", "*", feed_id="*", user_id="*") + self.personalization = AsyncPersonalization(self, token) + + token = self.create_jwt_token("reactions", "*", feed_id="*") + self.reactions = AsyncReactions(self, token) + + token = self.create_jwt_token("users", "*", feed_id="*") + self.users = AsyncUsers(self, token) + + def feed(self, feed_slug, user_id): + feed_slug = validate_feed_slug(feed_slug) + user_id = validate_user_id(user_id) + token = self.create_jwt_token("feed", "*", feed_id="*") + return AsyncFeed(self, feed_slug, user_id, token) + + async def put(self, *args, **kwargs): + return await self._make_request("PUT", *args, **kwargs) + + async def post(self, *args, **kwargs): + return await self._make_request("POST", *args, **kwargs) + + async def get(self, *args, **kwargs): + return await self._make_request("GET", *args, **kwargs) + + async def delete(self, *args, **kwargs): + return await self._make_request("DELETE", *args, **kwargs) + + async def add_to_many(self, activity, feeds): + data = {"activity": activity, "feeds": feeds} + token = self.create_jwt_token("feed", "*", feed_id="*") + return await self.post("feed/add_to_many/", token, data=data) + + async def follow_many(self, follows, activity_copy_limit=None): + params = None + + if activity_copy_limit is not None: + params = dict(activity_copy_limit=activity_copy_limit) + token = self.create_jwt_token("follower", "*", feed_id="*") + return await self.post("follow_many/", token, params=params, data=follows) + + async def unfollow_many(self, unfollows): + params = None + + token = self.create_jwt_token("follower", "*", feed_id="*") + return await self.post("unfollow_many/", token, params=params, data=unfollows) + + async def update_activities(self, activities): + if not isinstance(activities, (list, tuple, set)): + raise TypeError("Activities parameter should be of type list") + + auth_token = self.create_jwt_token("activities", "*", feed_id="*") + data = dict(activities=activities) + return await self.post("activities/", auth_token, data=data) + + async def update_activity(self, activity): + return await self.update_activities([activity]) + + async def get_activities( + self, ids=None, foreign_id_times=None, enrich=False, reactions=None, **params + ): + auth_token = self.create_jwt_token("activities", "*", feed_id="*") + + if ids is None and foreign_id_times is None: + raise TypeError( + "One the parameters ids or foreign_id_time must be provided and not None" + ) + + if ids is not None and foreign_id_times is not None: + raise TypeError( + "At most one of the parameters ids or foreign_id_time must be provided" + ) + + endpoint = "activities/" + if enrich or reactions is not None: + endpoint = "enrich/" + endpoint + + query_params = {**params} + + if ids is not None: + query_params["ids"] = ",".join(ids) + + if foreign_id_times is not None: + validate_foreign_id_time(foreign_id_times) + foreign_ids, timestamps = zip(*foreign_id_times) + timestamps = map(_datetime_encoder, timestamps) + query_params["foreign_ids"] = ",".join(foreign_ids) + query_params["timestamps"] = ",".join(timestamps) + + query_params.update(get_reaction_params(reactions)) + + return await self.get(endpoint, auth_token, params=query_params) + + async def activity_partial_update( + self, id=None, foreign_id=None, time=None, set=None, unset=None + ): + if id is None and (foreign_id is None or time is None): + raise TypeError( + "The id or foreign_id+time parameters must be provided and not be None" + ) + if id is not None and (foreign_id is not None or time is not None): + raise TypeError( + "Only one of the id or the foreign_id+time parameters can be provided" + ) + + data = {"set": set or {}, "unset": unset or []} + + if id is not None: + data["id"] = id + else: + data["foreign_id"] = foreign_id + data["time"] = time + + return await self.activities_partial_update(updates=[data]) + + async def activities_partial_update(self, updates=None): + auth_token = self.create_jwt_token("activities", "*", feed_id="*") + + data = {"changes": updates or []} + + return await self.post("activity/", auth_token, data=data) + + async def track_engagements(self, engagements): + auth_token = self.create_jwt_token("*", "*", feed_id="*") + await self.post( + "engagement/", + auth_token, + data={"content_list": engagements}, + service_name="analytics", + ) + + async def track_impressions(self, impressions): + auth_token = self.create_jwt_token("*", "*", feed_id="*") + await self.post( + "impression/", auth_token, data=impressions, service_name="analytics" + ) + + async def og(self, target_url): + auth_token = self.create_jwt_token("*", "*", feed_id="*") + params = {"url": target_url} + return await self.get("og/", auth_token, params=params) + + async def follow_stats(self, feed_id, followers_slugs=None, following_slugs=None): + auth_token = self.create_jwt_token("*", "*", feed_id="*") + params = {"followers": feed_id, "following": feed_id} + + if followers_slugs: + params["followers_slugs"] = ( + ",".join(followers_slugs) + if isinstance(followers_slugs, list) + else followers_slugs + ) + + if following_slugs: + params["following_slugs"] = ( + ",".join(following_slugs) + if isinstance(following_slugs, list) + else following_slugs + ) + + return await self.get("stats/follow/", auth_token, params=params) + + async def _make_request( + self, + method, + relative_url, + signature, + service_name="api", + params=None, + data=None, + ): + params = params or {} + data = data or {} + serialized = None + default_params = self.get_default_params() + params = self._check_params(params) + default_params.update(params) + headers = self.get_default_header() + headers["Authorization"] = signature + headers["stream-auth-type"] = "jwt" + + if not relative_url.endswith("/"): + relative_url += "/" + + url = self.get_full_url(service_name, relative_url) + + if method.lower() in ["post", "put", "delete"]: + serialized = serializer.dumps(data) + + async with self.session() as session: + async with session.request( + method, + url, + data=serialized, + headers=headers, + params=default_params, + timeout=self.timeout, + verify_ssl=False, + ) as response: + + logger.debug( + f"stream api call {response}, headers {headers} data {data}", + ) + return await self._parse_response(response) + + async def _parse_response(self, response): + try: + parsed_result = serializer.loads(await response.text()) + except (ValueError, ClientConnectionError): + parsed_result = None + if ( + parsed_result is None + or parsed_result.get("exception") + or response.status >= 500 + ): + self.raise_exception(parsed_result, status_code=response.status) + + return parsed_result + + def _check_params(self, params): + """There is no standard for boolean representation of boolean values in YARL""" + if not isinstance(params, dict): + raise TypeError("Invalid params type") + + for key, value in params.items(): + if isinstance(value, bool): + params[key] = str(value) + + return params diff --git a/stream/client.py b/stream/client/base.py similarity index 56% rename from stream/client.py rename to stream/client/base.py index 7eb4be8..e6a45ab 100644 --- a/stream/client.py +++ b/stream/client/base.py @@ -1,110 +1,21 @@ import json -import logging import os +from abc import ABC, abstractmethod -import jwt import requests -from requests import Request - -from stream import exceptions, serializer -from stream.collections import Collections -from stream.feed import Feed -from stream.personalization import Personalization -from stream.reactions import Reactions -from stream.serializer import _datetime_encoder -from stream.users import Users -from stream.utils import ( - validate_feed_slug, - validate_foreign_id_time, - validate_user_id, - get_reaction_params, -) + +from stream import exceptions try: from urllib.parse import urlparse except ImportError: from urlparse import urlparse -logger = logging.getLogger(__name__) - - -class StreamClient: - def __init__( - self, - api_key, - api_secret, - app_id, - version="v1.0", - timeout=6.0, - base_url=None, - location=None, - ): - """ - Initialize the client with the given api key and secret - - :param api_key: the api key - :param api_secret: the api secret - :param app_id: the app id - - **Example usage**:: - - import stream - # initialize the client - client = stream.connect('key', 'secret') - # get a feed object - feed = client.feed('aggregated:1') - # write data to the feed - activity_data = {'actor': 1, 'verb': 'tweet', 'object': 1} - activity_id = feed.add_activity(activity_data)['id'] - activities = feed.get() - - feed.follow('flat:3') - activities = feed.get() - feed.unfollow('flat:3') - feed.remove_activity(activity_id) - """ - self.api_key = api_key - self.api_secret = api_secret - self.app_id = app_id - self.version = version - self.timeout = timeout - self.location = location - - self.base_domain_name = "stream-io-api.com" - self.api_location = location - self.custom_api_port = None - self.protocol = "https" - - if os.environ.get("LOCAL"): - self.base_domain_name = "localhost" - self.protocol = "http" - self.custom_api_port = 8000 - self.timeout = 20 - elif base_url is not None: - parsed_url = urlparse(base_url) - self.base_domain_name = parsed_url.hostname - self.protocol = parsed_url.scheme - self.custom_api_port = parsed_url.port - self.api_location = "" - elif location is not None: - self.location = location - - self.base_analytics_url = "https://analytics.stream-io-api.com/analytics/" - - self.session = requests.Session() - - token = self.create_jwt_token("personalization", "*", feed_id="*", user_id="*") - self.personalization = Personalization(self, token) - - token = self.create_jwt_token("collections", "*", feed_id="*", user_id="*") - self.collections = Collections(self, token) - - token = self.create_jwt_token("reactions", "*", feed_id="*") - self.reactions = Reactions(self, token) +import jwt - token = self.create_jwt_token("users", "*", feed_id="*") - self.users = Users(self, token) +class AbstractStreamClient(ABC): + @abstractmethod def feed(self, feed_slug, user_id): """ Returns a Feed object @@ -112,193 +23,81 @@ def feed(self, feed_slug, user_id): :param feed_slug: the slug of the feed :param user_id: the user id """ - feed_slug = validate_feed_slug(feed_slug) - user_id = validate_user_id(user_id) - token = self.create_jwt_token("feed", "*", feed_id="*") - return Feed(self, feed_slug, user_id, token) + pass + @abstractmethod def get_default_params(self): """ Returns the params with the API key present """ - params = dict(api_key=self.api_key) - return params + pass + @abstractmethod def get_default_header(self): - base_headers = { - "Content-type": "application/json", - "X-Stream-Client": self.get_user_agent(), - } - return base_headers + pass + @abstractmethod def get_full_url(self, service_name, relative_url): - if self.api_location: - hostname = "%s%s.%s" % ( - self.api_location, - "" if service_name == "analytics" else f"-{service_name}", - self.base_domain_name, - ) - elif service_name: - hostname = "%s.%s" % (service_name, self.base_domain_name) - else: - hostname = self.base_domain_name - - if self.base_domain_name == "localhost": - hostname = "localhost" - - base_url = "%s://%s" % (self.protocol, hostname) - - if self.custom_api_port: - base_url = "%s:%s" % (base_url, self.custom_api_port) - - url = base_url + "/" + service_name + "/" + self.version + "/" + relative_url - return url + pass + @abstractmethod def get_user_agent(self): - from stream import __version__ - - agent = "stream-python-client-%s" % __version__ - return agent - - def _parse_response(self, response): - try: - parsed_result = serializer.loads(response.text) - except ValueError: - parsed_result = None - if ( - parsed_result is None - or parsed_result.get("exception") - or response.status_code >= 500 - ): - self.raise_exception(parsed_result, status_code=response.status_code) - return parsed_result + pass + @abstractmethod def create_user_token(self, user_id, **extra_data): """ Setup the payload for the given user_id with optional extra data (key, value pairs) and encode it using jwt """ - payload = {"user_id": user_id} - for k, v in extra_data.items(): - payload[k] = v - return jwt.encode(payload, self.api_secret, algorithm="HS256") + pass + @abstractmethod def create_jwt_token(self, resource, action, feed_id=None, user_id=None, **params): """ - Setup the payload for the given resource, action, feed or user + Set up the payload for the given resource, action, feed or user and encode it using jwt """ - payload = {**params, "action": action, "resource": resource} - if feed_id is not None: - payload["feed_id"] = feed_id - if user_id is not None: - payload["user_id"] = user_id - return jwt.encode(payload, self.api_secret, algorithm="HS256") - - def _make_request( - self, - method, - relative_url, - signature, - service_name="api", - params=None, - data=None, - ): - params = params or {} - data = data or {} - serialized = None - default_params = self.get_default_params() - default_params.update(params) - headers = self.get_default_header() - headers["Authorization"] = signature - headers["stream-auth-type"] = "jwt" - - if not relative_url.endswith("/"): - relative_url += "/" - - url = self.get_full_url(service_name, relative_url) - - if method.__name__ in ["post", "put", "delete"]: - serialized = serializer.dumps(data) - response = method( - url, - data=serialized, - headers=headers, - params=default_params, - timeout=self.timeout, - ) - logger.debug( - "stream api call %s, headers %s data %s", response.url, headers, data - ) - return self._parse_response(response) + pass + @abstractmethod def raise_exception(self, result, status_code): """ Map the exception code to an exception class and raise it If result.exception and result.detail are available use that Otherwise just raise a generic error """ - from stream.exceptions import get_exception_dict - - exception_class = exceptions.StreamApiException - - def errors_from_fields(exception_fields): - result = [] - if not isinstance(exception_fields, dict): - return exception_fields - - for field, errors in exception_fields.items(): - result.append('Field "%s" errors: %s' % (field, repr(errors))) - return result - - if result is not None: - error_message = result["detail"] - exception_fields = result.get("exception_fields") - if exception_fields is not None: - if isinstance(exception_fields, list): - errors = [ - errors_from_fields(exception_dict) - for exception_dict in exception_fields - ] - errors = [item for sublist in errors for item in sublist] - else: - errors = errors_from_fields(exception_fields) - - error_message = "\n".join(errors) - error_code = result.get("code") - exception_dict = get_exception_dict() - exception_class = exception_dict.get( - error_code, exceptions.StreamApiException - ) - else: - error_message = "GetStreamAPI%s" % status_code - exception = exception_class(error_message, status_code=status_code) - raise exception + pass + @abstractmethod def put(self, *args, **kwargs): """ Shortcut for make request """ - return self._make_request(self.session.put, *args, **kwargs) + pass + @abstractmethod def post(self, *args, **kwargs): """ Shortcut for make request """ - return self._make_request(self.session.post, *args, **kwargs) + pass + @abstractmethod def get(self, *args, **kwargs): """ Shortcut for make request """ - return self._make_request(self.session.get, *args, **kwargs) + pass + @abstractmethod def delete(self, *args, **kwargs): """ Shortcut for make request """ - return self._make_request(self.session.delete, *args, **kwargs) + pass + @abstractmethod def add_to_many(self, activity, feeds): """ Adds an activity to many feeds @@ -307,10 +106,9 @@ def add_to_many(self, activity, feeds): :param feeds: the list of follows (eg. ['feed:1', 'feed:2']) """ - data = {"activity": activity, "feeds": feeds} - token = self.create_jwt_token("feed", "*", feed_id="*") - return self.post("feed/add_to_many/", token, data=data) + pass + @abstractmethod def follow_many(self, follows, activity_copy_limit=None): """ Creates many follows @@ -319,13 +117,9 @@ def follow_many(self, follows, activity_copy_limit=None): eg. [{'source': source, 'target': target}] """ - params = None - - if activity_copy_limit is not None: - params = dict(activity_copy_limit=activity_copy_limit) - token = self.create_jwt_token("follower", "*", feed_id="*") - return self.post("follow_many/", token, params=params, data=follows) + pass + @abstractmethod def unfollow_many(self, unfollows): """ Unfollows many feeds at batch @@ -333,28 +127,23 @@ def unfollow_many(self, unfollows): eg. [{'source': source, 'target': target, 'keep_history': keep_history}] """ - params = None - - token = self.create_jwt_token("follower", "*", feed_id="*") - return self.post("unfollow_many/", token, params=params, data=unfollows) + pass + @abstractmethod def update_activities(self, activities): """ Update or create activities """ - if not isinstance(activities, (list, tuple, set)): - raise TypeError("Activities parameter should be of type list") - - auth_token = self.create_jwt_token("activities", "*", feed_id="*") - data = dict(activities=activities) - return self.post("activities/", auth_token, data=data) + pass + @abstractmethod def update_activity(self, activity): """ Update a single activity """ - return self.update_activities([activity]) + pass + @abstractmethod def get_activities( self, ids=None, foreign_id_times=None, enrich=False, reactions=None, **params ): @@ -366,38 +155,9 @@ def get_activities( ids: list of activity IDs foreign_id_time: list of tuples (foreign_id, time) """ - auth_token = self.create_jwt_token("activities", "*", feed_id="*") - - if ids is None and foreign_id_times is None: - raise TypeError( - "One the parameters ids or foreign_id_time must be provided and not None" - ) - - if ids is not None and foreign_id_times is not None: - raise TypeError( - "At most one of the parameters ids or foreign_id_time must be provided" - ) - - endpoint = "activities/" - if enrich or reactions is not None: - endpoint = "enrich/" + endpoint - - query_params = {**params} - - if ids is not None: - query_params["ids"] = ",".join(ids) - - if foreign_id_times is not None: - validate_foreign_id_time(foreign_id_times) - foreign_ids, timestamps = zip(*foreign_id_times) - timestamps = map(_datetime_encoder, timestamps) - query_params["foreign_ids"] = ",".join(foreign_ids) - query_params["timestamps"] = ",".join(timestamps) - - query_params.update(get_reaction_params(reactions)) - - return self.get(endpoint, auth_token, params=query_params) + pass + @abstractmethod def activity_partial_update( self, id=None, foreign_id=None, time=None, set=None, unset=None ): @@ -410,26 +170,9 @@ def activity_partial_update( set: object containing the set operations unset: list of unset operations """ + pass - if id is None and (foreign_id is None or time is None): - raise TypeError( - "The id or foreign_id+time parameters must be provided and not be None" - ) - if id is not None and (foreign_id is not None or time is not None): - raise TypeError( - "Only one of the id or the foreign_id+time parameters can be provided" - ) - - data = {"set": set or {}, "unset": unset or []} - - if id is not None: - data["id"] = id - else: - data["foreign_id"] = foreign_id - data["time"] = time - - return self.activities_partial_update(updates=[data]) - + @abstractmethod def activities_partial_update(self, updates=None): """ Partial update activity, via activity ID or Foreign ID + timestamp @@ -451,36 +194,18 @@ def activities_partial_update(self, updates=None): } ] """ + pass - auth_token = self.create_jwt_token("activities", "*", feed_id="*") - - data = {"changes": updates or []} - - return self.post("activity/", auth_token, data=data) - + @abstractmethod def create_redirect_url(self, target_url, user_id, events): """ Creates a redirect url for tracking the given events in the context of an email using Stream's analytics platform. Learn more at getstream.io/personalization """ - # generate the JWT token - auth_token = self.create_jwt_token( - "redirect_and_track", "*", "*", user_id=user_id - ) - # setup the params - params = dict(auth_type="jwt", authorization=auth_token, url=target_url) - params["api_key"] = self.api_key - params["events"] = json.dumps(events) - url = f"{self.base_analytics_url}redirect/" - # we get the url from the prepare request, this skips issues with - # python's urlencode implementation - request = Request("GET", url, params=params) - prepared_request = request.prepare() - # validate the target url is valid - Request("GET", target_url).prepare() - return prepared_request.url + pass + @abstractmethod def track_engagements(self, engagements): """ Creates a list of engagements @@ -515,15 +240,9 @@ def track_engagements(self, engagements): }, ] """ + pass - auth_token = self.create_jwt_token("*", "*", feed_id="*") - self.post( - "engagement/", - auth_token, - data={"content_list": engagements}, - service_name="analytics", - ) - + @abstractmethod def track_impressions(self, impressions): """ Creates a list of impressions @@ -547,46 +266,216 @@ def track_impressions(self, impressions): }, ] """ + pass - auth_token = self.create_jwt_token("*", "*", feed_id="*") - self.post("impression/", auth_token, data=impressions, service_name="analytics") - + @abstractmethod def og(self, target_url): """ Retrieve open graph information from a URL which you can then use to add images and a description to activities. """ - auth_token = self.create_jwt_token("*", "*", feed_id="*") - params = {"url": target_url} - return self.get("og/", auth_token, params=params) + pass + @abstractmethod def follow_stats(self, feed_id, followers_slugs=None, following_slugs=None): """ Retrieve the number of follower and following feed stats of a given feed. For each count, feed slugs can be provided to filter counts accordingly. eg. - client.follow_stats(me, followers_slugs=['user'], following_slugs=['commodities']) - this means to find counts of users following me and count of commodities I am following + client.follow_stats( + me, followers_slugs=['user'], following_slugs=['commodities'] + ) + this means to find counts of users following me and count + of commodities I am following """ - auth_token = self.create_jwt_token("*", "*", feed_id="*") - params = { - "followers": feed_id, - "following": feed_id, - } + pass + + @abstractmethod + def _make_request( + self, + method, + relative_url, + signature, + service_name="api", + params=None, + data=None, + ): + pass + + @abstractmethod + def _parse_response(self, response): + pass + + +class BaseStreamClient(AbstractStreamClient, ABC): + """ + Initialize the client with the given api key and secret + + :param api_key: the api key + :param api_secret: the api secret + :param app_id: the app id + + **Example usage**:: + + import stream + # initialize the client + client = stream.connect('key', 'secret') + # get a feed object + feed = client.feed('aggregated:1') + # write data to the feed + activity_data = {'actor': 1, 'verb': 'tweet', 'object': 1} + activity_id = feed.add_activity(activity_data)['id'] + activities = feed.get() + + feed.follow('flat:3') + activities = feed.get() + feed.unfollow('flat:3') + feed.remove_activity(activity_id) + """ + + def __init__( + self, + api_key, + api_secret, + app_id, + version="v1.0", + timeout=6.0, + base_url=None, + location=None, + ): + self.api_key = api_key + self.api_secret = api_secret + self.app_id = app_id + self.version = version + self.timeout = timeout + self.location = location + self.base_domain_name = "stream-io-api.com" + self.api_location = location + self.custom_api_port = None + self.protocol = "https" + + if os.environ.get("LOCAL"): + self.base_domain_name = "localhost" + self.protocol = "http" + self.custom_api_port = 8000 + self.timeout = 20 + elif base_url is not None: + parsed_url = urlparse(base_url) + self.base_domain_name = parsed_url.hostname + self.protocol = parsed_url.scheme + self.custom_api_port = parsed_url.port + self.api_location = "" + elif location is not None: + self.location = location + + self.base_analytics_url = "https://analytics.stream-io-api.com/analytics/" + + def create_user_token(self, user_id, **extra_data): + payload = {"user_id": user_id} + for k, v in extra_data.items(): + payload[k] = v + return jwt.encode(payload, self.api_secret, algorithm="HS256") + + def create_jwt_token(self, resource, action, feed_id=None, user_id=None, **params): + payload = {**params, "action": action, "resource": resource} + if feed_id is not None: + payload["feed_id"] = feed_id + if user_id is not None: + payload["user_id"] = user_id + return jwt.encode(payload, self.api_secret, algorithm="HS256") + + def raise_exception(self, result, status_code): + from stream.exceptions import get_exception_dict + + exception_class = exceptions.StreamApiException + + def errors_from_fields(exception_fields): + result = [] + if not isinstance(exception_fields, dict): + return exception_fields - if followers_slugs: - params["followers_slugs"] = ( - ",".join(followers_slugs) - if isinstance(followers_slugs, list) - else followers_slugs + for field, errors in exception_fields.items(): + result.append(f'Field "{field}" errors: {repr(errors)}') + return result + + if result is not None: + error_message = result["detail"] + exception_fields = result.get("exception_fields") + if exception_fields is not None: + if isinstance(exception_fields, list): + errors = [ + errors_from_fields(exception_dict) + for exception_dict in exception_fields + ] + errors = [item for sublist in errors for item in sublist] + else: + errors = errors_from_fields(exception_fields) + + error_message = "\n".join(errors) + error_code = result.get("code") + exception_dict = get_exception_dict() + exception_class = exception_dict.get( + error_code, exceptions.StreamApiException ) + else: + error_message = f"GetStreamAPI{status_code}" + exception = exception_class(error_message, status_code=status_code) + raise exception + + def create_redirect_url(self, target_url, user_id, events): + # generate the JWT token + auth_token = self.create_jwt_token( + "redirect_and_track", "*", "*", user_id=user_id + ) + # setup the params + params = dict(auth_type="jwt", authorization=auth_token, url=target_url) + params["api_key"] = self.api_key + params["events"] = json.dumps(events) + url = f"{self.base_analytics_url}redirect/" + # we get the url from the prepare request, this skips issues with + # python's urlencode implementation + request = requests.Request("GET", url, params=params) + prepared_request = request.prepare() + # validate the target url is valid + requests.Request("GET", target_url).prepare() + return prepared_request.url - if following_slugs: - params["following_slugs"] = ( - ",".join(following_slugs) - if isinstance(following_slugs, list) - else following_slugs + def get_full_url(self, service_name, relative_url): + if self.api_location: + hostname = "{}{}.{}".format( + self.api_location, + "" if service_name == "analytics" else f"-{service_name}", + self.base_domain_name, ) + elif service_name: + hostname = f"{service_name}.{self.base_domain_name}" + else: + hostname = self.base_domain_name + + if self.base_domain_name == "localhost": + hostname = "localhost" + + base_url = f"{self.protocol}://{hostname}" + + if self.custom_api_port: + base_url = f"{base_url}:{self.custom_api_port}" + + url = base_url + "/" + service_name + "/" + self.version + "/" + relative_url + return url + + def get_default_params(self): + params = dict(api_key=self.api_key) + return params + + def get_default_header(self): + base_headers = { + "Content-type": "application/json", + "X-Stream-Client": self.get_user_agent(), + } + return base_headers + + def get_user_agent(self): + from stream import __version__ - return self.get("stats/follow/", auth_token, params=params) + return f"stream-python-client-{__version__}" diff --git a/stream/client/client.py b/stream/client/client.py new file mode 100644 index 0000000..e52534e --- /dev/null +++ b/stream/client/client.py @@ -0,0 +1,289 @@ +import json +import logging + +import requests +from requests import Request + +from stream import serializer +from stream.client.base import BaseStreamClient +from stream.collections.collections import Collections +from stream.feed import Feed +from stream.personalization import Personalization +from stream.reactions import Reactions +from stream.serializer import _datetime_encoder +from stream.users import Users +from stream.utils import ( + get_reaction_params, + validate_feed_slug, + validate_foreign_id_time, + validate_user_id, +) + +try: + from urllib.parse import urlparse +except ImportError: + pass + # from urlparse import urlparse + +logger = logging.getLogger(__name__) + + +class StreamClient(BaseStreamClient): + def __init__( + self, + api_key, + api_secret, + app_id, + version="v1.0", + timeout=6.0, + base_url=None, + location=None, + ): + super().__init__( + api_key, + api_secret, + app_id, + version=version, + timeout=timeout, + base_url=base_url, + location=location, + ) + + self.session = requests.Session() + + token = self.create_jwt_token("personalization", "*", feed_id="*", user_id="*") + self.personalization = Personalization(self, token) + + token = self.create_jwt_token("collections", "*", feed_id="*", user_id="*") + self.collections = Collections(self, token) + + token = self.create_jwt_token("reactions", "*", feed_id="*") + self.reactions = Reactions(self, token) + + token = self.create_jwt_token("users", "*", feed_id="*") + self.users = Users(self, token) + + def feed(self, feed_slug, user_id): + feed_slug = validate_feed_slug(feed_slug) + user_id = validate_user_id(user_id) + token = self.create_jwt_token("feed", "*", feed_id="*") + return Feed(self, feed_slug, user_id, token) + + def put(self, *args, **kwargs): + return self._make_request(self.session.put, *args, **kwargs) + + def post(self, *args, **kwargs): + return self._make_request(self.session.post, *args, **kwargs) + + def get(self, *args, **kwargs): + return self._make_request(self.session.get, *args, **kwargs) + + def delete(self, *args, **kwargs): + return self._make_request(self.session.delete, *args, **kwargs) + + def add_to_many(self, activity, feeds): + data = {"activity": activity, "feeds": feeds} + token = self.create_jwt_token("feed", "*", feed_id="*") + return self.post("feed/add_to_many/", token, data=data) + + def follow_many(self, follows, activity_copy_limit=None): + params = None + + if activity_copy_limit is not None: + params = dict(activity_copy_limit=activity_copy_limit) + token = self.create_jwt_token("follower", "*", feed_id="*") + return self.post("follow_many/", token, params=params, data=follows) + + def unfollow_many(self, unfollows): + params = None + + token = self.create_jwt_token("follower", "*", feed_id="*") + return self.post("unfollow_many/", token, params=params, data=unfollows) + + def update_activities(self, activities): + if not isinstance(activities, (list, tuple, set)): + raise TypeError("Activities parameter should be of type list") + + auth_token = self.create_jwt_token("activities", "*", feed_id="*") + data = dict(activities=activities) + return self.post("activities/", auth_token, data=data) + + def update_activity(self, activity): + return self.update_activities([activity]) + + def get_activities( + self, ids=None, foreign_id_times=None, enrich=False, reactions=None, **params + ): + auth_token = self.create_jwt_token("activities", "*", feed_id="*") + + if ids is None and foreign_id_times is None: + raise TypeError( + "One the parameters ids or foreign_id_time must be provided and not None" + ) + + if ids is not None and foreign_id_times is not None: + raise TypeError( + "At most one of the parameters ids or foreign_id_time must be provided" + ) + + endpoint = "activities/" + if enrich or reactions is not None: + endpoint = "enrich/" + endpoint + + query_params = {**params} + + if ids is not None: + query_params["ids"] = ",".join(ids) + + if foreign_id_times is not None: + validate_foreign_id_time(foreign_id_times) + foreign_ids, timestamps = zip(*foreign_id_times) + timestamps = map(_datetime_encoder, timestamps) + query_params["foreign_ids"] = ",".join(foreign_ids) + query_params["timestamps"] = ",".join(timestamps) + + query_params.update(get_reaction_params(reactions)) + + return self.get(endpoint, auth_token, params=query_params) + + def activity_partial_update( + self, id=None, foreign_id=None, time=None, set=None, unset=None + ): + if id is None and (foreign_id is None or time is None): + raise TypeError( + "The id or foreign_id+time parameters must be provided and not be None" + ) + if id is not None and (foreign_id is not None or time is not None): + raise TypeError( + "Only one of the id or the foreign_id+time parameters can be provided" + ) + + data = {"set": set or {}, "unset": unset or []} + + if id is not None: + data["id"] = id + else: + data["foreign_id"] = foreign_id + data["time"] = time + + return self.activities_partial_update(updates=[data]) + + def activities_partial_update(self, updates=None): + + auth_token = self.create_jwt_token("activities", "*", feed_id="*") + + data = {"changes": updates or []} + + return self.post("activity/", auth_token, data=data) + + def create_redirect_url(self, target_url, user_id, events): + # generate the JWT token + auth_token = self.create_jwt_token( + "redirect_and_track", "*", "*", user_id=user_id + ) + # setup the params + params = dict(auth_type="jwt", authorization=auth_token, url=target_url) + params["api_key"] = self.api_key + params["events"] = json.dumps(events) + url = f"{self.base_analytics_url}redirect/" + # we get the url from the prepare request, this skips issues with + # python's urlencode implementation + request = Request("GET", url, params=params) + prepared_request = request.prepare() + # validate the target url is valid + Request("GET", target_url).prepare() + return prepared_request.url + + def track_engagements(self, engagements): + + auth_token = self.create_jwt_token("*", "*", feed_id="*") + self.post( + "engagement/", + auth_token, + data={"content_list": engagements}, + service_name="analytics", + ) + + def track_impressions(self, impressions): + + auth_token = self.create_jwt_token("*", "*", feed_id="*") + self.post("impression/", auth_token, data=impressions, service_name="analytics") + + def og(self, target_url): + auth_token = self.create_jwt_token("*", "*", feed_id="*") + params = {"url": target_url} + return self.get("og/", auth_token, params=params) + + def follow_stats(self, feed_id, followers_slugs=None, following_slugs=None): + auth_token = self.create_jwt_token("*", "*", feed_id="*") + params = { + "followers": feed_id, + "following": feed_id, + } + + if followers_slugs: + params["followers_slugs"] = ( + ",".join(followers_slugs) + if isinstance(followers_slugs, list) + else followers_slugs + ) + + if following_slugs: + params["following_slugs"] = ( + ",".join(following_slugs) + if isinstance(following_slugs, list) + else following_slugs + ) + + return self.get("stats/follow/", auth_token, params=params) + + def _make_request( + self, + method, + relative_url, + signature, + service_name="api", + params=None, + data=None, + ): + params = params or {} + data = data or {} + serialized = None + default_params = self.get_default_params() + default_params.update(params) + headers = self.get_default_header() + headers["Authorization"] = signature + headers["stream-auth-type"] = "jwt" + + if not relative_url.endswith("/"): + relative_url += "/" + + url = self.get_full_url(service_name, relative_url) + + if method.__name__ in ["post", "put", "delete"]: + serialized = serializer.dumps(data) + response = method( + url, + data=serialized, + headers=headers, + params=default_params, + timeout=self.timeout, + ) + logger.debug( + f"stream api call {response.url}, headers {headers} data {data}" + ) + return self._parse_response(response) + + def _parse_response(self, response): + try: + parsed_result = serializer.loads(response.text) + except ValueError: + parsed_result = None + if ( + parsed_result is None + or parsed_result.get("exception") + or response.status_code >= 500 + ): + self.raise_exception(parsed_result, status_code=response.status_code) + + return parsed_result diff --git a/stream/collections.py b/stream/collections.py deleted file mode 100644 index 967d5c6..0000000 --- a/stream/collections.py +++ /dev/null @@ -1,128 +0,0 @@ -class Collections: - def __init__(self, client, token): - """ - Used to manipulate data at the 'meta' endpoint - :param client: the api client - :param token: the token - """ - - self.client = client - self.token = token - - def create_reference(self, collection_name=None, id=None, entry=None): - if isinstance(entry, (dict,)): - _collection = entry["collection"] - _id = entry["id"] - elif collection_name is not None and id is not None: - _collection = collection_name - _id = id - else: - raise ValueError( - "must call with collection_name and id or with entry arguments" - ) - return "SO:%s:%s" % (_collection, _id) - - def upsert(self, collection_name, data): - """ - "Insert new or update existing data. - :param collection_name: Collection Name i.e 'user' - :param data: list of dictionaries - :return: http response, 201 if successful along with data posted. - - **Example**:: - client.collections.upsert('user', [{"id": '1', "name": "Juniper", "hobbies": ["Playing", "Sleeping", "Eating"]}, - {"id": '2', "name": "Ruby", "interests": ["Sunbeams", "Surprise Attacks"]}]) - """ - - if not isinstance(data, list): - data = [data] - - data_json = {collection_name: data} - - return self.client.post( - "collections/", - service_name="api", - signature=self.token, - data={"data": data_json}, - ) - - def select(self, collection_name, ids): - """ - Retrieve data from meta endpoint, can include data you've uploaded or personalization/analytic data - created by the stream team. - :param collection_name: Collection Name i.e 'user' - :param ids: list of ids of feed group i.e [123,456] - :return: meta data as json blob - - **Example**:: - client.collections.select('user', 1) - client.collections.select('user', [1,2,3]) - """ - - if not isinstance(ids, list): - ids = [ids] - - foreign_ids = ",".join( - "%s:%s" % (collection_name, k) for i, k in enumerate(ids) - ) - - return self.client.get( - "collections/", - service_name="api", - params={"foreign_ids": foreign_ids}, - signature=self.token, - ) - - def delete_many(self, collection_name, ids): - """ - Delete data from meta. - :param collection_name: Collection Name i.e 'user' - :param ids: list of ids to delete i.e [123,456] - :return: data that was deleted if successful or not. - - **Example**:: - client.collections.delete('user', '1') - client.collections.delete('user', ['1','2','3']) - """ - - if not isinstance(ids, list): - ids = [ids] - ids = [str(i) for i in ids] - - params = {"collection_name": collection_name, "ids": ids} - - return self.client.delete( - "collections/", service_name="api", params=params, signature=self.token - ) - - def add(self, collection_name, data, id=None, user_id=None): - payload = dict(id=id, data=data, user_id=user_id) - return self.client.post( - "collections/%s" % collection_name, - service_name="api", - signature=self.token, - data=payload, - ) - - def get(self, collection_name, id): - return self.client.get( - "collections/%s/%s" % (collection_name, id), - service_name="api", - signature=self.token, - ) - - def update(self, collection_name, id, data=None): - payload = dict(data=data) - return self.client.put( - "collections/%s/%s" % (collection_name, id), - service_name="api", - signature=self.token, - data=payload, - ) - - def delete(self, collection_name, id): - return self.client.delete( - "collections/%s/%s" % (collection_name, id), - service_name="api", - signature=self.token, - ) diff --git a/stream/collections/__init__.py b/stream/collections/__init__.py new file mode 100644 index 0000000..8264c83 --- /dev/null +++ b/stream/collections/__init__.py @@ -0,0 +1 @@ +from .collections import AsyncCollections, Collections diff --git a/stream/collections/base.py b/stream/collections/base.py new file mode 100644 index 0000000..44e091b --- /dev/null +++ b/stream/collections/base.py @@ -0,0 +1,100 @@ +from abc import ABC, abstractmethod + + +class AbstractCollection(ABC): + @abstractmethod + def create_reference(self, collection_name=None, id=None, entry=None): + pass + + @abstractmethod + def upsert(self, collection_name, data): + """ + "Insert new or update existing data. + :param collection_name: Collection Name i.e 'user' + :param data: list of dictionaries + :return: http response, 201 if successful along with data posted. + + **Example**:: + client.collections.upsert( + 'user', [ + {"id": '1', "name": "Juniper", "hobbies": ["Playing", "Sleeping", "Eating"]}, + {"id": '2', "name": "Ruby", "interests": ["Sunbeams", "Surprise Attacks"]} + ] + ) + """ + pass + + @abstractmethod + def select(self, collection_name, ids): + """ + Retrieve data from meta endpoint, can include data you've uploaded or + personalization/analytic data + created by the stream team. + :param collection_name: Collection Name i.e 'user' + :param ids: list of ids of feed group i.e [123,456] + :return: meta data as json blob + + **Example**:: + client.collections.select('user', 1) + client.collections.select('user', [1,2,3]) + """ + pass + + @abstractmethod + def delete_many(self, collection_name, ids): + """ + Delete data from meta. + :param collection_name: Collection Name i.e 'user' + :param ids: list of ids to delete i.e [123,456] + :return: data that was deleted if successful or not. + + **Example**:: + client.collections.delete('user', '1') + client.collections.delete('user', ['1','2','3']) + """ + pass + + @abstractmethod + def add(self, collection_name, data, id=None, user_id=None): + pass + + @abstractmethod + def get(self, collection_name, id): + pass + + @abstractmethod + def update(self, collection_name, id, data=None): + pass + + @abstractmethod + def delete(self, collection_name, id): + pass + + +class BaseCollection(AbstractCollection, ABC): + + URL = "collections/" + SERVICE_NAME = "api" + + def __init__(self, client, token): + """ + Used to manipulate data at the 'meta' endpoint + :param client: the api client + :param token: the token + """ + + self.client = client + self.token = token + + def create_reference(self, collection_name=None, id=None, entry=None): + if isinstance(entry, dict): + _collection = entry["collection"] + _id = entry["id"] + elif collection_name is not None and id is not None: + _collection = collection_name + _id = id + else: + raise ValueError( + "must call with collection_name and id or with entry arguments" + ) + return f"SO:{_collection}:{_id}" diff --git a/stream/collections/collections.py b/stream/collections/collections.py new file mode 100644 index 0000000..477062b --- /dev/null +++ b/stream/collections/collections.py @@ -0,0 +1,150 @@ +from stream.collections.base import BaseCollection + + +class Collections(BaseCollection): + def upsert(self, collection_name, data): + if not isinstance(data, list): + data = [data] + + data_json = {collection_name: data} + + return self.client.post( + self.URL, + service_name=self.SERVICE_NAME, + signature=self.token, + data={"data": data_json}, + ) + + def select(self, collection_name, ids): + if not isinstance(ids, list): + ids = [ids] + + foreign_ids = ",".join( + f"{collection_name}:{k}" for i, k in enumerate(ids) + ) + + return self.client.get( + self.URL, + service_name=self.SERVICE_NAME, + params={"foreign_ids": foreign_ids}, + signature=self.token, + ) + + def delete_many(self, collection_name, ids): + if not isinstance(ids, list): + ids = [ids] + ids = [str(i) for i in ids] + + params = {"collection_name": collection_name, "ids": ids} + + return self.client.delete( + self.URL, + service_name=self.SERVICE_NAME, + params=params, + signature=self.token, + ) + + def add(self, collection_name, data, id=None, user_id=None): + payload = dict(id=id, data=data, user_id=user_id) + return self.client.post( + f"{self.URL}/{collection_name}", + service_name=self.SERVICE_NAME, + signature=self.token, + data=payload, + ) + + def get(self, collection_name, id): + return self.client.get( + f"{self.URL}/{collection_name}/{id}", + service_name=self.SERVICE_NAME, + signature=self.token, + ) + + def update(self, collection_name, id, data=None): + payload = dict(data=data) + return self.client.put( + f"{self.URL}/{collection_name}/{id}", + service_name=self.SERVICE_NAME, + signature=self.token, + data=payload, + ) + + def delete(self, collection_name, id): + return self.client.delete( + f"{self.URL}/{collection_name}/{id}", + service_name=self.SERVICE_NAME, + signature=self.token, + ) + + +class AsyncCollections(BaseCollection): + async def upsert(self, collection_name, data): + if not isinstance(data, list): + data = [data] + + data_json = {collection_name: data} + + return await self.client.post( + self.URL, + service_name=self.SERVICE_NAME, + signature=self.token, + data={"data": data_json}, + ) + + async def select(self, collection_name, ids): + if not isinstance(ids, list): + ids = [ids] + + foreign_ids = ",".join(f"{collection_name}:{k}" for i, k in enumerate(ids)) + + return await self.client.get( + self.URL, + service_name=self.SERVICE_NAME, + params={"foreign_ids": foreign_ids}, + signature=self.token, + ) + + async def delete_many(self, collection_name, ids): + if not isinstance(ids, list): + ids = [ids] + ids = [str(i) for i in ids] + + params = {"collection_name": collection_name, "ids": ids} + return await self.client.delete( + self.URL, + service_name=self.SERVICE_NAME, + params=params, + signature=self.token, + ) + + async def get(self, collection_name, id): + return await self.client.get( + f"{self.URL}/{collection_name}/{id}", + service_name=self.SERVICE_NAME, + signature=self.token, + ) + + async def add(self, collection_name, data, id=None, user_id=None): + payload = dict(id=id, data=data, user_id=user_id) + return await self.client.post( + f"{self.URL}/{collection_name}", + service_name=self.SERVICE_NAME, + signature=self.token, + data=payload, + ) + + async def update(self, collection_name, id, data=None): + payload = dict(data=data) + return await self.client.put( + f"{self.URL}/{collection_name}/{id}", + service_name=self.SERVICE_NAME, + signature=self.token, + data=payload, + ) + + async def delete(self, collection_name, id): + return await self.client.delete( + f"{self.URL}/{collection_name}/{id}", + service_name=self.SERVICE_NAME, + signature=self.token, + ) diff --git a/stream/exceptions.py b/stream/exceptions.py index cd35f2c..6b1f70a 100644 --- a/stream/exceptions.py +++ b/stream/exceptions.py @@ -8,10 +8,10 @@ def __init__(self, error_message, status_code=None): code = 1 def __repr__(self): - return "%s (%s)" % (self.__class__.__name__, self.detail) + return f"{self.__class__.__name__} ({self.detail})" def __unicode__(self): - return "%s (%s)" % (self.__class__.__name__, self.detail) + return f"{self.__class__.__name__} ({self.detail})" class ApiKeyException(StreamApiException): diff --git a/stream/feed.py b/stream/feed.py deleted file mode 100644 index 3ef9317..0000000 --- a/stream/feed.py +++ /dev/null @@ -1,233 +0,0 @@ -from stream.utils import ( - validate_feed_id, - validate_feed_slug, - validate_user_id, - get_reaction_params, -) - - -class Feed: - def __init__(self, client, feed_slug, user_id, token): - """ - Initializes the Feed class - - :param client: the api client - :param slug: the slug of the feed, ie user, flat, notification - :param user_id: the id of the user - :param token: the token - """ - self.client = client - self.slug = feed_slug - self.user_id = str(user_id) - self.id = "%s:%s" % (feed_slug, user_id) - self.token = token.decode("utf-8") if isinstance(token, bytes) else token - - self.feed_url = "feed/%s/" % self.id.replace(":", "/") - self.enriched_feed_url = "enrich/feed/%s/" % self.id.replace(":", "/") - self.feed_targets_url = "feed_targets/%s/" % self.id.replace(":", "/") - self.feed_together = self.id.replace(":", "") - self.signature = self.feed_together + " " + self.token - - def create_scope_token(self, resource, action): - """ - creates the JWT token to perform an action on a owned resource - """ - return self.client.create_jwt_token( - resource, action, feed_id=self.feed_together - ) - - def get_readonly_token(self): - """ - creates the JWT token to perform readonly operations - """ - return self.create_scope_token("*", "read") - - def add_activity(self, activity_data): - """ - Adds an activity to the feed, this will also trigger an update - to all the feeds which follow this feed - - :param activity_data: a dict with the activity data - - **Example**:: - - activity_data = {'actor': 1, 'verb': 'tweet', 'object': 1} - activity_id = feed.add_activity(activity_data) - """ - if activity_data.get("to") and not isinstance( - activity_data.get("to"), (list, tuple, set) - ): - raise TypeError( - "please provide the activity's to field as a list not a string" - ) - - if activity_data.get("to"): - activity_data = activity_data.copy() - activity_data["to"] = self.add_to_signature(activity_data["to"]) - - token = self.create_scope_token("feed", "write") - return self.client.post(self.feed_url, data=activity_data, signature=token) - - def add_activities(self, activity_list): - """ - Adds a list of activities to the feed - - :param activity_list: a list with the activity data dicts - - **Example**:: - - activity_data = [ - {'actor': 1, 'verb': 'tweet', 'object': 1}, - {'actor': 2, 'verb': 'watch', 'object': 2}, - ] - result = feed.add_activities(activity_data) - """ - activities = [] - for activity_data in activity_list: - activity_data = activity_data.copy() - activities.append(activity_data) - if activity_data.get("to"): - activity_data["to"] = self.add_to_signature(activity_data["to"]) - token = self.create_scope_token("feed", "write") - data = dict(activities=activities) - if activities: - return self.client.post(self.feed_url, data=data, signature=token) - return None - - def remove_activity(self, activity_id=None, foreign_id=None): - """ - Removes an activity from the feed - - :param activity_id: the activity id to remove from this feed - (note this will also remove the activity from feeds which follow this feed) - :param foreign_id: the foreign id you provided when adding the activity - """ - identifier = activity_id or foreign_id - if not identifier: - raise ValueError("please either provide activity_id or foreign_id") - url = self.feed_url + "%s/" % identifier - params = dict() - token = self.create_scope_token("feed", "delete") - if foreign_id is not None: - params["foreign_id"] = "1" - return self.client.delete(url, signature=token, params=params) - - def get(self, enrich=False, reactions=None, **params): - """ - Get the activities in this feed - - **Example**:: - - # fast pagination using id filtering - feed.get(limit=10, id_lte=100292310) - - # slow pagination using offset - feed.get(limit=10, offset=10) - """ - for field in ["mark_read", "mark_seen"]: - value = params.get(field) - if isinstance(value, (list, tuple)): - params[field] = ",".join(value) - token = self.create_scope_token("feed", "read") - - if enrich or reactions is not None: - feed_url = self.enriched_feed_url - else: - feed_url = self.feed_url - - params.update(get_reaction_params(reactions)) - return self.client.get(feed_url, params=params, signature=token) - - def follow( - self, target_feed_slug, target_user_id, activity_copy_limit=None, **extra_data - ): - """ - Follows the given feed - - :param activity_copy_limit: how many activities should be copied from target feed - :param target_feed_slug: the slug of the target feed - :param target_user_id: the user id - """ - target_feed_slug = validate_feed_slug(target_feed_slug) - target_user_id = validate_user_id(target_user_id) - target_feed_id = "%s:%s" % (target_feed_slug, target_user_id) - url = self.feed_url + "follows/" - data = { - "target": target_feed_id, - "target_token": self.client.feed(target_feed_slug, target_user_id).token, - } - if activity_copy_limit is not None: - data["activity_copy_limit"] = activity_copy_limit - token = self.create_scope_token("follower", "write") - data.update(extra_data) - return self.client.post(url, data=data, signature=token) - - def unfollow(self, target_feed_slug, target_user_id, keep_history=False): - """ - Unfollow the given feed - """ - target_feed_slug = validate_feed_slug(target_feed_slug) - target_user_id = validate_user_id(target_user_id) - target_feed_id = "%s:%s" % (target_feed_slug, target_user_id) - token = self.create_scope_token("follower", "delete") - url = self.feed_url + "follows/%s/" % target_feed_id - params = {} - if keep_history: - params["keep_history"] = True - return self.client.delete(url, signature=token, params=params) - - def followers(self, offset=0, limit=25, feeds=None): - """ - Lists the followers for the given feed - """ - feeds = ",".join(feeds) if feeds is not None else "" - params = {"limit": limit, "offset": offset, "filter": feeds} - url = self.feed_url + "followers/" - token = self.create_scope_token("follower", "read") - return self.client.get(url, params=params, signature=token) - - def following(self, offset=0, limit=25, feeds=None): - """ - List the feeds which this feed is following - """ - feeds = ",".join(feeds) if feeds is not None else "" - params = {"offset": offset, "limit": limit, "filter": feeds} - url = self.feed_url + "follows/" - token = self.create_scope_token("follower", "read") - return self.client.get(url, params=params, signature=token) - - def add_to_signature(self, recipients): - """ - Takes a list of recipients such as ['user:1', 'user:2'] - and turns it into a list with the tokens included - ['user:1 token', 'user:2 token'] - """ - data = [] - for recipient in recipients: - validate_feed_id(recipient) - feed_slug, user_id = recipient.split(":") - feed = self.client.feed(feed_slug, user_id) - data.append("%s %s" % (recipient, feed.token)) - return data - - def update_activity_to_targets( - self, - foreign_id, - time, - new_targets=None, - added_targets=None, - removed_targets=None, - ): - data = {"foreign_id": foreign_id, "time": time} - - if new_targets is not None: - data["new_targets"] = new_targets - if added_targets is not None: - data["added_targets"] = added_targets - if removed_targets is not None: - data["removed_targets"] = removed_targets - - url = self.feed_targets_url + "activity_to_targets/" - - token = self.create_scope_token("feed_targets", "write") - return self.client.post(url, data=data, signature=token) diff --git a/stream/feed/__init__.py b/stream/feed/__init__.py new file mode 100644 index 0000000..1f3c784 --- /dev/null +++ b/stream/feed/__init__.py @@ -0,0 +1 @@ +from .feeds import AsyncFeed, Feed diff --git a/stream/feed/base.py b/stream/feed/base.py new file mode 100644 index 0000000..dc76fce --- /dev/null +++ b/stream/feed/base.py @@ -0,0 +1,172 @@ +from abc import ABC, abstractmethod + +from stream.utils import validate_feed_id + + +class AbstractFeed(ABC): + @abstractmethod + def create_scope_token(self, resource, action): + """ + creates the JWT token to perform an action on a owned resource + """ + pass + + @abstractmethod + def get_readonly_token(self): + """ + creates the JWT token to perform readonly operations + """ + pass + + @abstractmethod + def add_activity(self, activity_data): + """ + Adds an activity to the feed, this will also trigger an update + to all the feeds which follow this feed + + :param activity_data: a dict with the activity data + + **Example**:: + + activity_data = {'actor': 1, 'verb': 'tweet', 'object': 1} + activity_id = feed.add_activity(activity_data) + """ + pass + + @abstractmethod + def add_activities(self, activity_list): + """ + Adds a list of activities to the feed + + :param activity_list: a list with the activity data dicts + + **Example**:: + + activity_data = [ + {'actor': 1, 'verb': 'tweet', 'object': 1}, + {'actor': 2, 'verb': 'watch', 'object': 2}, + ] + result = feed.add_activities(activity_data) + """ + pass + + @abstractmethod + def remove_activity(self, activity_id=None, foreign_id=None): + """ + Removes an activity from the feed + + :param activity_id: the activity id to remove from this feed + (note this will also remove the activity from feeds which follow this feed) + :param foreign_id: the foreign id you provided when adding the activity + """ + pass + + @abstractmethod + def get(self, enrich=False, reactions=None, **params): + """ + Get the activities in this feed + + **Example**:: + + # fast pagination using id filtering + feed.get(limit=10, id_lte=100292310) + + # slow pagination using offset + feed.get(limit=10, offset=10) + """ + pass + + @abstractmethod + def follow( + self, target_feed_slug, target_user_id, activity_copy_limit=None, **extra_data + ): + """ + Follows the given feed + + :param activity_copy_limit: how many activities should be copied from target + feed + :param target_feed_slug: the slug of the target feed + :param target_user_id: the user id + """ + pass + + @abstractmethod + def unfollow(self, target_feed_slug, target_user_id, keep_history=False): + """ + Unfollow the given feed + """ + pass + + @abstractmethod + def followers(self, offset=0, limit=25, feeds=None): + """ + Lists the followers for the given feed + """ + pass + + @abstractmethod + def following(self, offset=0, limit=25, feeds=None): + """ + List the feeds which this feed is following + """ + pass + + @abstractmethod + def add_to_signature(self, recipients): + """ + Takes a list of recipients such as ['user:1', 'user:2'] + and turns it into a list with the tokens included + ['user:1 token', 'user:2 token'] + """ + pass + + @abstractmethod + def update_activity_to_targets( + self, + foreign_id, + time, + new_targets=None, + added_targets=None, + removed_targets=None, + ): + pass + + +class BaseFeed(AbstractFeed, ABC): + def __init__(self, client, feed_slug, user_id, token): + """ + Initializes the Feed class + + :param client: the api client + :param feed_slug: the slug of the feed, ie user, flat, notification + :param user_id: the id of the user + :param token: the token + """ + self.client = client + self.slug = feed_slug + self.user_id = f"{user_id}" + self.id = f"{feed_slug}:{user_id}" + self.token = token.decode("utf-8") if isinstance(token, bytes) else token + _id = self.id.replace(":", "/") + self.feed_url = f"feed/{_id}/" + self.enriched_feed_url = f"enrich/feed/{_id}/" + self.feed_targets_url = f"feed_targets/{_id}/" + self.feed_together = self.id.replace(":", "") + self.signature = f"{self.feed_together} {self.token}" + + def create_scope_token(self, resource, action): + return self.client.create_jwt_token( + resource, action, feed_id=self.feed_together + ) + + def get_readonly_token(self): + return self.create_scope_token("*", "read") + + def add_to_signature(self, recipients): + data = [] + for recipient in recipients: + validate_feed_id(recipient) + feed_slug, user_id = recipient.split(":") + feed = self.client.feed(feed_slug, user_id) + data.append(f"{recipient} {feed.token}") + return data diff --git a/stream/feed/feeds.py b/stream/feed/feeds.py new file mode 100644 index 0000000..5305427 --- /dev/null +++ b/stream/feed/feeds.py @@ -0,0 +1,238 @@ +from stream.feed.base import BaseFeed +from stream.utils import get_reaction_params, validate_feed_slug, validate_user_id + + +class Feed(BaseFeed): + def add_activity(self, activity_data): + if activity_data.get("to") and not isinstance( + activity_data.get("to"), (list, tuple, set) + ): + raise TypeError( + "please provide the activity's to field as a list not a string" + ) + + if activity_data.get("to"): + activity_data = activity_data.copy() + activity_data["to"] = self.add_to_signature(activity_data["to"]) + + token = self.create_scope_token("feed", "write") + return self.client.post(self.feed_url, data=activity_data, signature=token) + + def add_activities(self, activity_list): + activities = [] + for activity_data in activity_list: + activity_data = activity_data.copy() + activities.append(activity_data) + if activity_data.get("to"): + activity_data["to"] = self.add_to_signature(activity_data["to"]) + token = self.create_scope_token("feed", "write") + data = dict(activities=activities) + if activities: + return self.client.post(self.feed_url, data=data, signature=token) + return None + + def remove_activity(self, activity_id=None, foreign_id=None): + identifier = activity_id or foreign_id + if not identifier: + raise ValueError("please either provide activity_id or foreign_id") + url = f"{self.feed_url}{identifier}/" + params = dict() + token = self.create_scope_token("feed", "delete") + if foreign_id is not None: + params["foreign_id"] = "1" + return self.client.delete(url, signature=token, params=params) + + def get(self, enrich=False, reactions=None, **params): + for field in ["mark_read", "mark_seen"]: + value = params.get(field) + if isinstance(value, (list, tuple)): + params[field] = ",".join(value) + token = self.create_scope_token("feed", "read") + + if enrich or reactions is not None: + feed_url = self.enriched_feed_url + else: + feed_url = self.feed_url + + params.update(get_reaction_params(reactions)) + return self.client.get(feed_url, params=params, signature=token) + + def follow( + self, target_feed_slug, target_user_id, activity_copy_limit=None, **extra_data + ): + target_feed_slug = validate_feed_slug(target_feed_slug) + target_user_id = validate_user_id(target_user_id) + target_feed_id = f"{target_feed_slug}:{target_user_id}" + url = f"{self.feed_url}follows/" + target_token = self.client.feed(target_feed_slug, target_user_id).token + data = {"target": target_feed_id, "target_token": target_token} + if activity_copy_limit is not None: + data["activity_copy_limit"] = activity_copy_limit + token = self.create_scope_token("follower", "write") + data.update(extra_data) + return self.client.post(url, data=data, signature=token) + + def unfollow(self, target_feed_slug, target_user_id, keep_history=False): + target_feed_slug = validate_feed_slug(target_feed_slug) + target_user_id = validate_user_id(target_user_id) + target_feed_id = f"{target_feed_slug}:{target_user_id}" + token = self.create_scope_token("follower", "delete") + url = f"{self.feed_url}follows/{target_feed_id}/" + params = {} + if keep_history: + params["keep_history"] = True + return self.client.delete(url, signature=token, params=params) + + def followers(self, offset=0, limit=25, feeds=None): + feeds = ",".join(feeds) if feeds is not None else "" + params = {"limit": limit, "offset": offset, "filter": feeds} + url = f"{self.feed_url}followers/" + token = self.create_scope_token("follower", "read") + return self.client.get(url, params=params, signature=token) + + def following(self, offset=0, limit=25, feeds=None): + feeds = ",".join(feeds) if feeds is not None else "" + params = {"offset": offset, "limit": limit, "filter": feeds} + url = f"{self.feed_url}follows/" + token = self.create_scope_token("follower", "read") + return self.client.get(url, params=params, signature=token) + + def update_activity_to_targets( + self, + foreign_id, + time, + new_targets=None, + added_targets=None, + removed_targets=None, + ): + data = {"foreign_id": foreign_id, "time": time} + + if new_targets is not None: + data["new_targets"] = new_targets + if added_targets is not None: + data["added_targets"] = added_targets + if removed_targets is not None: + data["removed_targets"] = removed_targets + + url = f"{self.feed_targets_url}activity_to_targets/" + token = self.create_scope_token("feed_targets", "write") + return self.client.post(url, data=data, signature=token) + + +class AsyncFeed(BaseFeed): + async def add_activity(self, activity_data): + if activity_data.get("to") and not isinstance( + activity_data.get("to"), (list, tuple, set) + ): + raise TypeError( + "please provide the activity's to field as a list not a string" + ) + + if activity_data.get("to"): + activity_data = activity_data.copy() + activity_data["to"] = self.add_to_signature(activity_data["to"]) + + token = self.create_scope_token("feed", "write") + return await self.client.post( + self.feed_url, data=activity_data, signature=token + ) + + async def add_activities(self, activity_list): + activities = [] + for activity_data in activity_list: + activity_data = activity_data.copy() + activities.append(activity_data) + if activity_data.get("to"): + activity_data["to"] = self.add_to_signature(activity_data["to"]) + token = self.create_scope_token("feed", "write") + data = dict(activities=activities) + if not activities: + return + + return await self.client.post(self.feed_url, data=data, signature=token) + + async def remove_activity(self, activity_id=None, foreign_id=None): + identifier = activity_id or foreign_id + if not identifier: + raise ValueError("please either provide activity_id or foreign_id") + url = f"{self.feed_url}{identifier}/" + params = dict() + token = self.create_scope_token("feed", "delete") + if foreign_id is not None: + params["foreign_id"] = "1" + return await self.client.delete(url, signature=token, params=params) + + async def get(self, enrich=False, reactions=None, **params): + for field in ["mark_read", "mark_seen"]: + value = params.get(field) + if isinstance(value, (list, tuple)): + params[field] = ",".join(value) + + token = self.create_scope_token("feed", "read") + if enrich or reactions is not None: + feed_url = self.enriched_feed_url + else: + feed_url = self.feed_url + + params.update(get_reaction_params(reactions)) + return await self.client.get(feed_url, params=params, signature=token) + + async def follow( + self, target_feed_slug, target_user_id, activity_copy_limit=None, **extra_data + ): + target_feed_slug = validate_feed_slug(target_feed_slug) + target_user_id = validate_user_id(target_user_id) + target_feed_id = f"{target_feed_slug}:{target_user_id}" + url = f"{self.feed_url}follows/" + target_token = self.client.feed(target_feed_slug, target_user_id).token + data = {"target": target_feed_id, "target_token": target_token} + if activity_copy_limit is not None: + data["activity_copy_limit"] = activity_copy_limit + token = self.create_scope_token("follower", "write") + data.update(extra_data) + return await self.client.post(url, data=data, signature=token) + + async def unfollow(self, target_feed_slug, target_user_id, keep_history=False): + target_feed_slug = validate_feed_slug(target_feed_slug) + target_user_id = validate_user_id(target_user_id) + target_feed_id = f"{target_feed_slug}:{target_user_id}" + token = self.create_scope_token("follower", "delete") + url = f"{self.feed_url}follows/{target_feed_id}/" + params = {} + if keep_history: + params["keep_history"] = True + return await self.client.delete(url, signature=token, params=params) + + async def followers(self, offset=0, limit=25, feeds=None): + feeds = ",".join(feeds) if feeds is not None else "" + params = {"limit": limit, "offset": offset, "filter": feeds} + url = f"{self.feed_url}followers/" + token = self.create_scope_token("follower", "read") + return await self.client.get(url, params=params, signature=token) + + async def following(self, offset=0, limit=25, feeds=None): + feeds = ",".join(feeds) if feeds is not None else "" + params = {"offset": offset, "limit": limit, "filter": feeds} + url = f"{self.feed_url}follows/" + token = self.create_scope_token("follower", "read") + return await self.client.get(url, params=params, signature=token) + + async def update_activity_to_targets( + self, + foreign_id, + time, + new_targets=None, + added_targets=None, + removed_targets=None, + ): + data = {"foreign_id": foreign_id, "time": time} + if new_targets is not None: + data["new_targets"] = new_targets + if added_targets is not None: + data["added_targets"] = added_targets + if removed_targets is not None: + data["removed_targets"] = removed_targets + + url = f"{self.feed_targets_url}activity_to_targets/" + token = self.create_scope_token("feed_targets", "write") + return await self.client.post(url, data=data, signature=token) diff --git a/stream/personalization.py b/stream/personalization.py deleted file mode 100644 index 77f3174..0000000 --- a/stream/personalization.py +++ /dev/null @@ -1,66 +0,0 @@ -class Personalization: - def __init__(self, client, token): - """ - Methods to interact with personalized feeds. - :param client: the api client - :param token: the token - """ - - self.client = client - self.token = token - - def get(self, resource, **params): - """ - Get personalized activities for this feed - :param resource: personalized resource endpoint i.e "follow_recommendations" - :param params: params to pass to url i.e user_id = "user:123" - :return: personalized feed - - **Example**:: - personalization.get('follow_recommendations', user_id=123, limit=10, offset=10) - """ - - return self.client.get( - resource, - service_name="personalization", - params=params, - signature=self.token, - ) - - def post(self, resource, **params): - """ - Generic function to post data to personalization endpoint - :param resource: personalized resource endpoint i.e "follow_recommendations" - :param params: params to pass to url (data is a reserved keyword to post to body) - - - **Example**:: - #Accept or reject recommendations. - personalization.post('follow_recommendations', user_id=123, accepted=[123,345], - rejected=[456]) - """ - - data = params["data"] or None - - return self.client.post( - resource, - service_name="personalization", - params=params, - signature=self.token, - data=data, - ) - - def delete(self, resource, **params): - """ - shortcut to delete metadata or activities - :param resource: personalized url endpoint typical "meta" - :param params: params to pass to url i.e user_id = "user:123" - :return: data that was deleted if successful or not. - """ - - return self.client.delete( - resource, - service_name="personalization", - params=params, - signature=self.token, - ) diff --git a/stream/personalization/__init__.py b/stream/personalization/__init__.py new file mode 100644 index 0000000..99dfa3b --- /dev/null +++ b/stream/personalization/__init__.py @@ -0,0 +1 @@ +from .personalizations import AsyncPersonalization, Personalization diff --git a/stream/personalization/base.py b/stream/personalization/base.py new file mode 100644 index 0000000..730d78f --- /dev/null +++ b/stream/personalization/base.py @@ -0,0 +1,30 @@ +from abc import ABC, abstractmethod + + +class AbstractPersonalization(ABC): + @abstractmethod + def get(self, resource, **params): + pass + + @abstractmethod + def post(self, resource, **params): + pass + + @abstractmethod + def delete(self, resource, **params): + pass + + +class BasePersonalization(AbstractPersonalization, ABC): + + SERVICE_NAME = "personalization" + + def __init__(self, client, token): + """ + Methods to interact with personalized feeds. + :param client: the api client + :param token: the token + """ + + self.client = client + self.token = token diff --git a/stream/personalization/personalizations.py b/stream/personalization/personalizations.py new file mode 100644 index 0000000..61d9909 --- /dev/null +++ b/stream/personalization/personalizations.py @@ -0,0 +1,117 @@ +from stream.personalization.base import BasePersonalization + + +class Personalization(BasePersonalization): + def get(self, resource, **params): + """ + Get personalized activities for this feed + :param resource: personalized resource endpoint i.e "follow_recommendations" + :param params: params to pass to url i.e user_id = "user:123" + :return: personalized feed + + **Example**:: + personalization.get('follow_recommendations', user_id=123, limit=10, offset=10) + """ + + return self.client.get( + resource, + service_name=self.SERVICE_NAME, + params=params, + signature=self.token, + ) + + def post(self, resource, **params): + """ + Generic function to post data to personalization endpoint + :param resource: personalized resource endpoint i.e "follow_recommendations" + :param params: params to pass to url (data is a reserved keyword to post to body) + + + **Example**:: + #Accept or reject recommendations. + personalization.post('follow_recommendations', user_id=123, accepted=[123,345], + rejected=[456]) + """ + + data = params["data"] or None + + return self.client.post( + resource, + service_name=self.SERVICE_NAME, + params=params, + signature=self.token, + data=data, + ) + + def delete(self, resource, **params): + """ + shortcut to delete metadata or activities + :param resource: personalized url endpoint typical "meta" + :param params: params to pass to url i.e user_id = "user:123" + :return: data that was deleted if successful or not. + """ + + return self.client.delete( + resource, + service_name=self.SERVICE_NAME, + params=params, + signature=self.token, + ) + + +class AsyncPersonalization(BasePersonalization): + async def get(self, resource, **params): + """ + Get personalized activities for this feed + :param resource: personalized resource endpoint i.e "follow_recommendations" + :param params: params to pass to url i.e user_id = "user:123" + :return: personalized feed + + **Example**:: + personalization.get('follow_recommendations', user_id=123, limit=10, offset=10) + """ + + return await self.client.get( + resource, + service_name=self.SERVICE_NAME, + params=params, + signature=self.token, + ) + + async def post(self, resource, **params): + """ + Generic function to post data to personalization endpoint + :param resource: personalized resource endpoint i.e "follow_recommendations" + :param params: params to pass to url (data is a reserved keyword to post to body) + + + **Example**:: + #Accept or reject recommendations. + personalization.post('follow_recommendations', user_id=123, accepted=[123,345], + rejected=[456]) + """ + + data = params["data"] or None + + return await self.client.post( + resource, + service_name=self.SERVICE_NAME, + params=params, + signature=self.token, + data=data, + ) + + async def delete(self, resource, **params): + """ + shortcut to delete metadata or activities + :param resource: personalized url endpoint typical "meta" + :param params: params to pass to url i.e user_id = "user:123" + :return: data that was deleted if successful or not. + """ + + return await self.client.delete( + resource, + service_name=self.SERVICE_NAME, + params=params, + signature=self.token, + ) diff --git a/stream/reactions.py b/stream/reactions.py deleted file mode 100644 index 88a60a4..0000000 --- a/stream/reactions.py +++ /dev/null @@ -1,89 +0,0 @@ -class Reactions: - def __init__(self, client, token): - self.client = client - self.token = token - - def add( - self, - kind, - activity_id, - user_id, - data=None, - target_feeds=None, - target_feeds_extra_data=None, - ): - payload = dict( - kind=kind, - activity_id=activity_id, - data=data, - target_feeds=target_feeds, - target_feeds_extra_data=target_feeds_extra_data, - user_id=user_id, - ) - return self.client.post( - "reaction/", service_name="api", signature=self.token, data=payload - ) - - def get(self, reaction_id): - return self.client.get( - "reaction/%s" % reaction_id, service_name="api", signature=self.token - ) - - def update(self, reaction_id, data=None, target_feeds=None): - payload = dict(data=data, target_feeds=target_feeds) - return self.client.put( - "reaction/%s" % reaction_id, - service_name="api", - signature=self.token, - data=payload, - ) - - def delete(self, reaction_id): - return self.client.delete( - "reaction/%s" % reaction_id, service_name="api", signature=self.token - ) - - def add_child( - self, - kind, - parent_id, - user_id, - data=None, - target_feeds=None, - target_feeds_extra_data=None, - ): - payload = dict( - kind=kind, - parent=parent_id, - data=data, - target_feeds=target_feeds, - target_feeds_extra_data=target_feeds_extra_data, - user_id=user_id, - ) - return self.client.post( - "reaction/", service_name="api", signature=self.token, data=payload - ) - - def filter(self, **params): - lookup_field = "" - lookup_value = "" - - kind = params.pop("kind", None) - - if "reaction_id" in params: - lookup_field = "reaction_id" - lookup_value = params.pop("reaction_id") - elif "activity_id" in params: - lookup_field = "activity_id" - lookup_value = params.pop("activity_id") - elif "user_id" in params: - lookup_field = "user_id" - lookup_value = params.pop("user_id") - - endpoint = "reaction/%s/%s/" % (lookup_field, lookup_value) - if kind is not None: - endpoint = "reaction/%s/%s/%s/" % (lookup_field, lookup_value, kind) - - return self.client.get( - endpoint, service_name="api", signature=self.token, params=params - ) diff --git a/stream/reactions/__init__.py b/stream/reactions/__init__.py new file mode 100644 index 0000000..e550051 --- /dev/null +++ b/stream/reactions/__init__.py @@ -0,0 +1 @@ +from .reaction import AsyncReactions, Reactions diff --git a/stream/reactions/base.py b/stream/reactions/base.py new file mode 100644 index 0000000..b83794e --- /dev/null +++ b/stream/reactions/base.py @@ -0,0 +1,75 @@ +from abc import ABC, abstractmethod + + +class AbstractReactions(ABC): + @abstractmethod + def add( + self, + kind, + activity_id, + user_id, + data=None, + target_feeds=None, + target_feeds_extra_data=None, + ): + pass + + @abstractmethod + def get(self, reaction_id): + pass + + @abstractmethod + def update(self, reaction_id, data=None, target_feeds=None): + pass + + @abstractmethod + def delete(self, reaction_id): + pass + + @abstractmethod + def add_child( + self, + kind, + parent_id, + user_id, + data=None, + target_feeds=None, + target_feeds_extra_data=None, + ): + pass + + @abstractmethod + def filter(self, **params): + pass + + +class BaseReactions(AbstractReactions, ABC): + + API_ENDPOINT = "reaction/" + SERVICE_NAME = "api" + + def __init__(self, client, token): + self.client = client + self.token = token + + def _prepare_endpoint_for_filter(self, **params): + lookup_field = "" + lookup_value = "" + + kind = params.pop("kind", None) + + if params.get("reaction_id"): + lookup_field = "reaction_id" + lookup_value = params.pop("reaction_id") + elif params.get("activity_id"): + lookup_field = "activity_id" + lookup_value = params.pop("activity_id") + elif params.get("user_id"): + lookup_field = "user_id" + lookup_value = params.pop("user_id") + + endpoint = f"{self.API_ENDPOINT}{lookup_field}/{lookup_value}/" + if kind is not None: + endpoint += f"{kind}/" + + return endpoint diff --git a/stream/reactions/reaction.py b/stream/reactions/reaction.py new file mode 100644 index 0000000..0466b21 --- /dev/null +++ b/stream/reactions/reaction.py @@ -0,0 +1,163 @@ +from stream.reactions.base import BaseReactions + + +class Reactions(BaseReactions): + def add( + self, + kind, + activity_id, + user_id, + data=None, + target_feeds=None, + target_feeds_extra_data=None, + ): + payload = dict( + kind=kind, + activity_id=activity_id, + data=data, + target_feeds=target_feeds, + target_feeds_extra_data=target_feeds_extra_data, + user_id=user_id, + ) + return self.client.post( + self.API_ENDPOINT, + service_name=self.SERVICE_NAME, + signature=self.token, + data=payload, + ) + + def get(self, reaction_id): + url = f"{self.API_ENDPOINT}{reaction_id}" + return self.client.get( + url, service_name=self.SERVICE_NAME, signature=self.token + ) + + def update(self, reaction_id, data=None, target_feeds=None): + payload = dict(data=data, target_feeds=target_feeds) + url = f"{self.API_ENDPOINT}{reaction_id}" + return self.client.put( + url, + service_name=self.SERVICE_NAME, + signature=self.token, + data=payload, + ) + + def delete(self, reaction_id): + url = f"{self.API_ENDPOINT}{reaction_id}" + return self.client.delete( + url, service_name=self.SERVICE_NAME, signature=self.token + ) + + def add_child( + self, + kind, + parent_id, + user_id, + data=None, + target_feeds=None, + target_feeds_extra_data=None, + ): + payload = dict( + kind=kind, + parent=parent_id, + data=data, + target_feeds=target_feeds, + target_feeds_extra_data=target_feeds_extra_data, + user_id=user_id, + ) + return self.client.post( + self.API_ENDPOINT, + service_name=self.SERVICE_NAME, + signature=self.token, + data=payload, + ) + + def filter(self, **params): + endpoint = self._prepare_endpoint_for_filter(**params) + return self.client.get( + endpoint, + service_name=self.SERVICE_NAME, + signature=self.token, + params=params, + ) + + +class AsyncReactions(BaseReactions): + async def add( + self, + kind, + activity_id, + user_id, + data=None, + target_feeds=None, + target_feeds_extra_data=None, + ): + payload = dict( + kind=kind, + activity_id=activity_id, + data=data, + target_feeds=target_feeds, + target_feeds_extra_data=target_feeds_extra_data, + user_id=user_id, + ) + return await self.client.post( + self.API_ENDPOINT, + service_name=self.SERVICE_NAME, + signature=self.token, + data=payload, + ) + + async def get(self, reaction_id): + url = f"{self.API_ENDPOINT}{reaction_id}" + return await self.client.get( + url, service_name=self.SERVICE_NAME, signature=self.token + ) + + async def update(self, reaction_id, data=None, target_feeds=None): + payload = dict(data=data, target_feeds=target_feeds) + url = f"{self.API_ENDPOINT}{reaction_id}" + return await self.client.put( + url, + service_name=self.SERVICE_NAME, + signature=self.token, + data=payload, + ) + + async def delete(self, reaction_id): + url = f"{self.API_ENDPOINT}{reaction_id}" + return await self.client.delete( + url, service_name=self.SERVICE_NAME, signature=self.token + ) + + async def add_child( + self, + kind, + parent_id, + user_id, + data=None, + target_feeds=None, + target_feeds_extra_data=None, + ): + payload = dict( + kind=kind, + parent=parent_id, + data=data, + target_feeds=target_feeds, + target_feeds_extra_data=target_feeds_extra_data, + user_id=user_id, + ) + return await self.client.post( + self.API_ENDPOINT, + service_name=self.SERVICE_NAME, + signature=self.token, + data=payload, + ) + + async def filter(self, **params): + endpoint = self._prepare_endpoint_for_filter(**params) + return await self.client.get( + endpoint, + service_name=self.SERVICE_NAME, + signature=self.token, + params=params, + ) diff --git a/stream/tests/conftest.py b/stream/tests/conftest.py new file mode 100644 index 0000000..af77ee4 --- /dev/null +++ b/stream/tests/conftest.py @@ -0,0 +1,60 @@ +import asyncio + +import pytest +from stream import connect + + +def wrapper(meth): + async def _parse_response(*args, **kwargs): + response = await meth(*args, **kwargs) + assert "duration" in response + return response + + return _parse_response + + +@pytest.fixture(scope="module") +def event_loop(request): + """Create an instance of the default event loop for each test case.""" + loop = asyncio.get_event_loop_policy().new_event_loop() + yield loop + loop.close() + + +@pytest.fixture +async def async_client(): + key = "n8udba92h9hf" + secret = "5mjaw2a5bynt3fzdnxy2pwxh7fw89gg2dq3mm4jy56vhrwnadfst7yssatmaxhkv" + client = connect(key, secret, location="qa", timeout=30, use_async=True) + wrapper(client._parse_response) + yield client + + +@pytest.fixture +def user1(async_client): + return async_client.feed("user", "1") + + +@pytest.fixture +def user2(async_client): + return async_client.feed("user", "2") + + +@pytest.fixture +def aggregated2(async_client): + return async_client.feed("aggregated", "2") + + +@pytest.fixture +def aggregated3(async_client): + return async_client.feed("aggregated", "3") + + +@pytest.fixture +def topic(async_client): + return async_client.feed("topic", "1") + + +@pytest.fixture +def flat3(async_client): + return async_client.feed("flat", "3") diff --git a/stream/tests/test_async_client.py b/stream/tests/test_async_client.py new file mode 100644 index 0000000..123b0f1 --- /dev/null +++ b/stream/tests/test_async_client.py @@ -0,0 +1,1342 @@ +import asyncio +import random +from datetime import datetime, timedelta +from uuid import uuid1, uuid4 + +import pytz +import pytest +from dateutil.tz import tzlocal + +import stream +from stream.exceptions import ApiKeyException, InputException +from stream.tests.test_client import get_unique_postfix + + +def assert_first_activity_id_equal(activities, correct_activity_id): + activity_id = None + if activities: + activity_id = activities[0]["id"] + assert activity_id == correct_activity_id + + +def assert_first_activity_id_not_equal(activities, correct_activity_id): + activity_id = None + if activities: + activity_id = activities[0]["id"] + assert activity_id != correct_activity_id + + +def _get_first_aggregated_activity(activities): + try: + return activities[0]["activities"][0] + except IndexError: + pass + + +def _get_first_activity(activities): + try: + return activities[0] + except IndexError: + pass + + +def assert_datetime_almost_equal(a, b): + difference = abs(a - b) + if difference > timedelta(milliseconds=1): + assert a == b + + +def assert_clearly_not_equal(a, b): + difference = abs(a - b) + if difference < timedelta(milliseconds=1): + raise ValueError("the dates are too close") + + +async def _test_sleep(production_wait): + """ + when testing against a live API, sometimes we need a small sleep to + ensure data stability, however when testing locally the wait does + not need to be as long + :param production_wait: float, number of seconds to sleep when hitting real API + :return: None + """ + sleep_time = production_wait + await asyncio.sleep(sleep_time) + + +@pytest.mark.asyncio +async def test_update_activities_create(async_client): + activities = [ + { + "actor": "user:1", + "verb": "do", + "object": "object:1", + "foreign_id": "object:1", + "time": datetime.utcnow().isoformat(), + } + ] + + response = await async_client.update_activities(activities) + assert response + + +@pytest.mark.asyncio +async def test_add_activity(async_client): + feed = async_client.feed("user", "py1") + activity_data = {"actor": 1, "verb": "tweet", "object": 1} + response = await feed.add_activity(activity_data) + activity_id = response["id"] + response = await feed.get(limit=1) + activities = response["results"] + assert activities[0]["id"] == activity_id + + +@pytest.mark.asyncio +async def test_add_activity_to_inplace_change(async_client): + feed = async_client.feed("user", "py1") + team_feed = async_client.feed("user", "teamy") + activity_data = {"actor": 1, "verb": "tweet", "object": 1} + activity_data["to"] = [team_feed.id] + await feed.add_activity(activity_data) + assert activity_data["to"] == [team_feed.id] + + +@pytest.mark.asyncio +async def test_add_activities_to_inplace_change(async_client): + feed = async_client.feed("user", "py1") + team_feed = async_client.feed("user", "teamy") + activity_data = {"actor": 1, "verb": "tweet", "object": 1} + activity_data["to"] = [team_feed.id] + await feed.add_activities([activity_data]) + assert activity_data["to"] == [team_feed.id] + + +@pytest.mark.asyncio +async def test_add_activity_to(async_client): + # test for sending an activities to the team feed using to + feeds = ["user", "teamy", "team_follower"] + user_feed, team_feed, team_follower_feed = map( + lambda x: async_client.feed("user", x), feeds + ) + await team_follower_feed.follow(team_feed.slug, team_feed.user_id) + activity_data = {"actor": 1, "verb": "tweet", "object": 1, "to": [team_feed.id]} + activity = await user_feed.add_activity(activity_data) + activity_id = activity["id"] + + # see if the new activity is also in the team feed + response = await team_feed.get(limit=1) + activities = response["results"] + assert activities[0]["id"] == activity_id + assert activities[0]["origin"] is None + # see if the fanout process also works + response = await team_follower_feed.get(limit=1) + activities = response["results"] + assert activities[0]["id"] == activity_id + assert activities[0]["origin"] == team_feed.id + # and validate removing also works + await user_feed.remove_activity(activity["id"]) + # check the user pyto feed + response = await team_feed.get(limit=1) + activities = response["results"] + assert_first_activity_id_not_equal(activities, activity_id) + # and the flat feed + response = await team_follower_feed.get(limit=1) + activities = response["results"] + assert_first_activity_id_not_equal(activities, activity_id) + + +@pytest.mark.asyncio +async def test_remove_activity(user1): + activity_data = {"actor": 1, "verb": "tweet", "object": 1} + activity = await user1.add_activity(activity_data) + activity_id = activity["id"] + response = await user1.get(limit=8) + activities = response["results"] + assert len(activities) == 1 + + await user1.remove_activity(activity_id) + # verify that no activities were returned + response = await user1.get(limit=8) + activities = response["results"] + assert len(activities) == 0 + + +@pytest.mark.asyncio +async def test_remove_activity_by_foreign_id(user1): + activity_data = { + "actor": 1, + "verb": "tweet", + "object": 1, + "foreign_id": "tweet:10", + } + + await user1.add_activity(activity_data) + response = await user1.get(limit=8) + activities = response["results"] + assert len(activities) == 1 + assert activities[0]["id"] != "" + assert activities[0]["foreign_id"] == "tweet:10" + + await user1.remove_activity(foreign_id="tweet:10") + # verify that no activities were returned + response = await user1.get(limit=8) + activities = response["results"] + assert len(activities) == 0 + # verify this doesn't raise an error, but fails silently + await user1.remove_activity(foreign_id="tweet:unknownandmissing") + + +@pytest.mark.asyncio +async def test_add_activities(user1): + activity_data = [ + {"actor": 1, "verb": "tweet", "object": 1}, + {"actor": 2, "verb": "watch", "object": 2}, + ] + response = await user1.add_activities(activity_data) + activity_ids = [a["id"] for a in response["activities"]] + response = await user1.get(limit=2) + activities = response["results"] + get_activity_ids = [a["id"] for a in activities] + assert get_activity_ids == activity_ids[::-1] + + +@pytest.mark.asyncio +async def test_add_activities_to(async_client, user1): + pyto2 = async_client.feed("user", "pyto2") + pyto3 = async_client.feed("user", "pyto3") + + to = [pyto2.id, pyto3.id] + activity_data = [ + {"actor": 1, "verb": "tweet", "object": 1, "to": to}, + {"actor": 2, "verb": "watch", "object": 2, "to": to}, + ] + response = await user1.add_activities(activity_data) + activity_ids = [a["id"] for a in response["activities"]] + response = await user1.get(limit=2) + activities = response["results"] + get_activity_ids = [a["id"] for a in activities] + assert get_activity_ids == activity_ids[::-1] + # test first target + response = await pyto2.get(limit=2) + activities = response["results"] + get_activity_ids = [a["id"] for a in activities] + assert get_activity_ids == activity_ids[::-1] + # test second target + response = await pyto3.get(limit=2) + activities = response["results"] + get_activity_ids = [a["id"] for a in activities] + assert get_activity_ids == activity_ids[::-1] + + +@pytest.mark.asyncio +async def test_follow_and_source(async_client): + feed = async_client.feed("user", "test_follow") + agg_feed = async_client.feed("aggregated", "test_follow") + actor_id = random.randint(10, 100000) + activity_data = {"actor": actor_id, "verb": "tweet", "object": 1} + response = await feed.add_activity(activity_data) + activity_id = response["id"] + await agg_feed.follow(feed.slug, feed.user_id) + + response = await agg_feed.get(limit=3) + activities = response["results"] + activity = _get_first_aggregated_activity(activities) + activity_id_found = activity["id"] if activity is not None else None + assert activity["origin"] == feed.id + assert activity_id_found == activity_id + + +@pytest.mark.asyncio +async def test_empty_followings(async_client): + asocial = async_client.feed("user", "asocialpython") + followings = await asocial.following() + assert followings["results"] == [] + + +@pytest.mark.asyncio +async def test_get_followings(async_client): + social = async_client.feed("user", "psocial") + await social.follow("user", "apy") + await social.follow("user", "bpy") + await social.follow("user", "cpy") + followings = await social.following(offset=0, limit=2) + assert len(followings["results"]) == 2 + assert followings["results"][0]["feed_id"] == social.id + assert followings["results"][0]["target_id"] == "user:cpy" + followings = await social.following(offset=1, limit=2) + assert len(followings["results"]) == 2 + assert followings["results"][0]["feed_id"] == social.id + assert followings["results"][0]["target_id"] == "user:bpy" + + +@pytest.mark.asyncio +async def test_empty_followers(async_client): + asocial = async_client.feed("user", "asocialpython") + followers = await asocial.followers() + assert followers["results"] == [] + + +@pytest.mark.asyncio +async def test_get_followers(async_client): + social = async_client.feed("user", "psocial") + spammy1 = async_client.feed("user", "spammy1") + spammy2 = async_client.feed("user", "spammy2") + spammy3 = async_client.feed("user", "spammy3") + for feed in [spammy1, spammy2, spammy3]: + await feed.follow("user", social.user_id) + followers = await social.followers(offset=0, limit=2) + assert len(followers["results"]) == 2 + assert followers["results"][0]["feed_id"] == spammy3.id + assert followers["results"][0]["target_id"] == social.id + followers = await social.followers(offset=1, limit=2) + assert len(followers["results"]) == 2 + assert followers["results"][0]["feed_id"] == spammy2.id + assert followers["results"][0]["target_id"] == social.id + + +@pytest.mark.asyncio +async def test_empty_do_i_follow(async_client): + social = async_client.feed("user", "psocial") + await social.follow("user", "apy") + await social.follow("user", "bpy") + followings = await social.following(feeds=["user:missingpy"]) + assert followings["results"] == [] + + +@pytest.mark.asyncio +async def test_do_i_follow(async_client): + social = async_client.feed("user", "psocial") + await social.follow("user", "apy") + await social.follow("user", "bpy") + followings = await social.following(feeds=["user:apy"]) + assert len(followings["results"]) == 1 + assert followings["results"][0]["feed_id"] == social.id + assert followings["results"][0]["target_id"] == "user:apy" + + +@pytest.mark.asyncio +async def test_update_activity_to_targets(user1): + now = datetime.utcnow().isoformat() + foreign_id = "user:1" + activity_data = { + "actor": 1, + "verb": "tweet", + "object": 1, + "foreign_id": foreign_id, + "time": now, + "to": ["user:1", "user:2"], + } + await user1.add_activity(activity_data) + + ret = await user1.update_activity_to_targets( + foreign_id, now, new_targets=["user:3", "user:2"] + ) + assert len(ret["activity"]["to"]) == 2 + assert "user:2" in ret["activity"]["to"] + assert "user:3" in ret["activity"]["to"] + + ret = await user1.update_activity_to_targets( + foreign_id, + now, + added_targets=["user:4", "user:5"], + removed_targets=["user:3"], + ) + assert len(ret["activity"]["to"]) == 3 + assert "user:2" in ret["activity"]["to"] + assert "user:4" in ret["activity"]["to"] + assert "user:5" in ret["activity"]["to"] + + +@pytest.mark.asyncio +async def test_get(user1): + activity_data = {"actor": 1, "verb": "tweet", "object": 1} + response1 = await user1.add_activity(activity_data) + activity_id = response1["id"] + activity_data = {"actor": 2, "verb": "add", "object": 2} + response2 = await user1.add_activity(activity_data) + activity_id_two = response2["id"] + activity_data = {"actor": 3, "verb": "watch", "object": 2} + response3 = await user1.add_activity(activity_data) + activity_id_three = response3["id"] + response = await user1.get(limit=2) + activities = response["results"] + # verify the first two results + assert len(activities) == 2 + assert activities[0]["id"] == activity_id_three + assert activities[1]["id"] == activity_id_two + # try offset based + response = await user1.get(limit=2, offset=1) + activities = response["results"] + assert activities[0]["id"] == activity_id_two + # try id_lt based + response = await user1.get(limit=2, id_lt=activity_id_two) + activities = response["results"] + assert activities[0]["id"] == activity_id + + +@pytest.mark.asyncio +async def test_get_not_marked_seen(async_client): + notification_feed = async_client.feed("notification", "test_mark_seen") + response = await notification_feed.get(limit=3) + activities = response["results"] + for activity in activities: + assert not activity["is_seen"] + + +@pytest.mark.asyncio +async def test_mark_seen_on_get(async_client): + notification_feed = async_client.feed("notification", "test_mark_seen") + response = await notification_feed.get(limit=100) + activities = response["results"] + for activity in activities: + await notification_feed.remove_activity(activity["id"]) + + old_activities = [ + await notification_feed.add_activity( + {"actor": 1, "verb": "tweet", "object": 1} + ), + await notification_feed.add_activity( + {"actor": 2, "verb": "add", "object": 2} + ), + await notification_feed.add_activity( + {"actor": 3, "verb": "watch", "object": 3} + ), + ] + + await notification_feed.get( + mark_seen=[old_activities[0]["id"], old_activities[1]["id"]] + ) + + response = await notification_feed.get(limit=3) + activities = response["results"] + + # is the seen state correct + for activity in activities: + # using a loop in case we're retrieving activities in a different + # order than old_activities + if old_activities[0]["id"] == activity["id"]: + assert activity["is_seen"] + if old_activities[1]["id"] == activity["id"]: + assert activity["is_seen"] + if old_activities[2]["id"] == activity["id"]: + assert not activity["is_seen"] + + # see if the state properly resets after we add another activity + await notification_feed.add_activity( + {"actor": 3, "verb": "watch", "object": 3} + ) # ['id'] + response = await notification_feed.get(limit=3) + activities = response["results"] + assert not activities[0]["is_seen"] + assert len(activities[0]["activities"]) == 2 + + +@pytest.mark.asyncio +async def test_mark_read_by_id(async_client): + notification_feed = async_client.feed("notification", "py2") + response = await notification_feed.get(limit=3) + activities = response["results"] + ids = [] + for activity in activities: + ids.append(activity["id"]) + assert not activity["is_read"] + ids = ids[:2] + await notification_feed.get(mark_read=ids) + response = await notification_feed.get(limit=3) + activities = response["results"] + for activity in activities: + if activity["id"] in ids: + assert activity["is_read"] + assert not activity["is_seen"] + + +@pytest.mark.asyncio +async def test_api_key_exception(): + client = stream.connect( + "5crf3bhfzesnMISSING", + "tfq2sdqpj9g446sbv653x3aqmgn33hsn8uzdc9jpskaw8mj6vsnhzswuwptuj9su", + use_async=True, + ) + user1 = client.feed("user", "1") + activity_data = { + "actor": 1, + "verb": "tweet", + "object": 1, + "debug_example_undefined": "test", + } + with pytest.raises(ApiKeyException): + await user1.add_activity(activity_data) + + +@pytest.mark.asyncio +async def test_complex_field(user1): + activity_data = { + "actor": 1, + "verb": "tweet", + "object": 1, + "participants": ["Tommaso", "Thierry"], + } + response = await user1.add_activity(activity_data) + activity_id = response["id"] + response = await user1.get(limit=1) + activities = response["results"] + assert activities[0]["id"] == activity_id + assert activities[0]["participants"] == ["Tommaso", "Thierry"] + + +@pytest.mark.asyncio +async def test_uniqueness(user1): + """ + In order for things to be considere unique they need: + a.) The same time and activity data + b.) The same time and foreign id + """ + + utcnow = datetime.now(tz=pytz.UTC) + activity_data = {"actor": 1, "verb": "tweet", "object": 1, "time": utcnow} + await user1.add_activity(activity_data) + await user1.add_activity(activity_data) + response = await user1.get(limit=2) + activities = response["results"] + assert_datetime_almost_equal(activities[0]["time"], utcnow) + if len(activities) > 1: + assert_clearly_not_equal(activities[1]["time"], utcnow) + + +@pytest.mark.asyncio +async def test_uniqueness_topic(flat3, topic, user1): + """ + In order for things to be considere unique they need: + a.) The same time and activity data, or + b.) The same time and foreign id + """ + # follow both the topic and the user + await flat3.follow("topic", topic.user_id) + await flat3.follow("user", user1.user_id) + # add the same activity twice + now = datetime.now(tzlocal()) + tweet = f"My Way {get_unique_postfix()}" + activity_data = { + "actor": 1, + "verb": "tweet", + "object": 1, + "time": now, + "tweet": tweet, + } + await topic.add_activity(activity_data) + await user1.add_activity(activity_data) + # verify that flat3 contains the activity exactly once + response = await flat3.get(limit=3) + activity_tweets = [a.get("tweet") for a in response["results"]] + assert activity_tweets.count(tweet) == 1 + + +@pytest.mark.asyncio +async def test_uniqueness_foreign_id(user1): + now = datetime.now(tzlocal()) + utcnow = now.astimezone(pytz.utc) + + activity_data = { + "actor": 1, + "verb": "tweet", + "object": 1, + "foreign_id": "tweet:11", + "time": utcnow, + } + await user1.add_activity(activity_data) + + activity_data = { + "actor": 2, + "verb": "tweet", + "object": 3, + "foreign_id": "tweet:11", + "time": utcnow, + } + await user1.add_activity(activity_data) + response = await user1.get(limit=10) + activities = response["results"] + # the second post should have overwritten the first one (because they + # had same id) + + assert len(activities) == 1 + assert activities[0]["object"] == "3" + assert activities[0]["foreign_id"] == "tweet:11" + assert_datetime_almost_equal(activities[0]["time"], utcnow) + + +@pytest.mark.asyncio +async def test_time_ordering(user2): + """ + datetime.datetime.now(tz=pytz.utc) is our recommended approach + so if we add an activity + add one using time + add another activity it should be in the right spot + """ + + # timedelta is used to "make sure" that ordering is known even though + # server time is not + custom_time = datetime.now(tz=pytz.utc) - timedelta(days=1) + + feed = user2 + for index, activity_time in enumerate([None, custom_time, None]): + await _test_sleep(1) # so times are a bit different + activity_data = { + "actor": 1, + "verb": "tweet", + "object": 1, + "foreign_id": f"tweet:{index}", + "time": activity_time, + } + await feed.add_activity(activity_data) + + response = await feed.get(limit=3) + activities = response["results"] + # the second post should have overwritten the first one (because they + # had same id) + assert activities[0]["foreign_id"] == "tweet:2" + assert activities[1]["foreign_id"] == "tweet:0" + assert activities[2]["foreign_id"] == "tweet:1" + assert_datetime_almost_equal(activities[2]["time"], custom_time) + + +@pytest.mark.asyncio +async def test_missing_actor(user1): + activity_data = { + "verb": "tweet", + "object": 1, + "debug_example_undefined": "test", + } + try: + await user1.add_activity(activity_data) + raise ValueError("should have raised InputException") + except InputException: + pass + + +@pytest.mark.asyncio +async def test_follow_many(async_client): + sources = [async_client.feed("user", str(i)).id for i in range(10)] + targets = [async_client.feed("flat", str(i)).id for i in range(10)] + feeds = [{"source": s, "target": t} for s, t in zip(sources, targets)] + await async_client.follow_many(feeds) + + for target in targets: + response = await async_client.feed(*target.split(":")).followers() + follows = response["results"] + assert len(follows) == 1 + assert follows[0]["feed_id"] in sources + assert follows[0]["target_id"] == target + + for source in sources: + response = await async_client.feed(*source.split(":")).following() + follows = response["results"] + assert len(follows) == 1 + assert follows[0]["feed_id"] in sources + assert follows[0]["target_id"] == source + + +@pytest.mark.asyncio +async def test_follow_many_acl(async_client): + sources = [async_client.feed("user", str(i)) for i in range(10)] + # ensure every source is empty first + for feed in sources: + response = await feed.get(limit=100) + activities = response["results"] + for activity in activities: + await feed.remove_activity(activity["id"]) + + targets = [async_client.feed("flat", str(i)) for i in range(10)] + # ensure every source is empty first + for feed in targets: + response = await feed.get(limit=100) + activities = response["results"] + for activity in activities: + await feed.remove_activity(activity["id"]) + # add activity to each target feed + activity = { + "actor": "barry", + "object": "09", + "verb": "tweet", + "time": datetime.utcnow().isoformat(), + } + for feed in targets: + await feed.add_activity(activity) + response = await feed.get(limit=5) + assert len(response["results"]) == 1 + + sources_id = [feed.id for feed in sources] + targets_id = [target.id for target in targets] + feeds = [{"source": s, "target": t} for s, t in zip(sources_id, targets_id)] + + await async_client.follow_many(feeds, activity_copy_limit=0) + + for feed in sources: + response = await feed.get(limit=5) + activities = response["results"] + assert len(activities) == 0 + + +@pytest.mark.asyncio +async def test_unfollow_many(async_client): + unfollows = [ + {"source": "user:1", "target": "timeline:1"}, + {"source": "user:2", "target": "timeline:2", "keep_history": False}, + ] + + await async_client.unfollow_many(unfollows) + unfollows.append({"source": "user:1", "target": 42}) + + async def failing_unfollow(): + await async_client.unfollow_many(unfollows) + + with pytest.raises(InputException): + await failing_unfollow() + + +@pytest.mark.asyncio +async def test_add_to_many(async_client): + activity = {"actor": 1, "verb": "tweet", "object": 1, "custom": "data"} + feeds = [async_client.feed("flat", str(i)).id for i in range(10, 20)] + await async_client.add_to_many(activity, feeds) + + for feed in feeds: + feed = async_client.feed(*feed.split(":")) + response = await feed.get() + assert response["results"][0]["custom"] == "data" + + +@pytest.mark.asyncio +async def test_get_activities_empty_ids(async_client): + response = await async_client.get_activities(ids=[str(uuid1())]) + assert len(response["results"]) == 0 + + +@pytest.mark.asyncio +async def test_get_activities_empty_foreign_ids(async_client): + response = await async_client.get_activities( + foreign_id_times=[("fid-x", datetime.utcnow())] + ) + assert len(response["results"]) == 0 + + +@pytest.mark.asyncio +async def test_get_activities_full(async_client): + dt = datetime.utcnow() + fid = "awesome-test" + + activity = { + "actor": "barry", + "object": "09", + "verb": "tweet", + "time": dt, + "foreign_id": fid, + } + + feed = async_client.feed("user", "test_get_activity") + response = await feed.add_activity(activity) + + response = await async_client.get_activities(ids=[response["id"]]) + assert len(response["results"]) == 1 + foreign_id = response["results"][0]["foreign_id"] + assert activity["foreign_id"] == foreign_id + + response = await async_client.get_activities(foreign_id_times=[(fid, dt)]) + assert len(response["results"]) == 1 + foreign_id = response["results"][0]["foreign_id"] + assert activity["foreign_id"] == foreign_id + + +@pytest.mark.asyncio +async def test_get_activities_full_with_enrichment(async_client): + dt = datetime.utcnow() + fid = "awesome-test" + + actor = await async_client.users.add(str(uuid1()), data={"name": "barry"}) + activity = { + "actor": async_client.users.create_reference(actor["id"]), + "object": "09", + "verb": "tweet", + "time": dt, + "foreign_id": fid, + } + + feed = async_client.feed("user", "test_get_activity") + activity = await feed.add_activity(activity) + + reaction1 = await async_client.reactions.add("like", activity["id"], "liker") + reaction2 = await async_client.reactions.add("reshare", activity["id"], "sharer") + + def validate(response): + assert len(response["results"]) == 1 + assert response["results"][0]["id"] == activity["id"] + assert response["results"][0]["foreign_id"] == activity["foreign_id"] + assert response["results"][0]["actor"]["data"]["name"] == "barry" + latest_reactions = response["results"][0]["latest_reactions"] + assert len(latest_reactions) == 2 + assert latest_reactions["like"][0]["id"] == reaction1["id"] + assert latest_reactions["reshare"][0]["id"] == reaction2["id"] + assert response["results"][0]["reaction_counts"] == {"like": 1, "reshare": 1} + + reactions = {"recent": True, "counts": True} + validate( + await async_client.get_activities(ids=[activity["id"]], reactions=reactions) + ) + validate( + await async_client.get_activities( + foreign_id_times=[(fid, dt)], reactions=reactions + ) + ) + + +@pytest.mark.asyncio +async def test_get_activities_full_with_enrichment_and_reaction_kinds(async_client): + dt = datetime.utcnow() + fid = "awesome-test" + + actor = await async_client.users.add(str(uuid1()), data={"name": "barry"}) + activity = { + "actor": async_client.users.create_reference(actor["id"]), + "object": "09", + "verb": "tweet", + "time": dt, + "foreign_id": fid, + } + + feed = async_client.feed("user", "test_get_activity") + activity = await feed.add_activity(activity) + + await async_client.reactions.add("like", activity["id"], "liker") + await async_client.reactions.add("reshare", activity["id"], "sharer") + await async_client.reactions.add("comment", activity["id"], "commenter") + + reactions = {"recent": True, "counts": True, "kinds": "like,comment"} + response = await async_client.get_activities( + ids=[activity["id"]], reactions=reactions + ) + assert len(response["results"]) == 1 + assert response["results"][0]["id"] == activity["id"] + assert sorted(response["results"][0]["latest_reactions"].keys()) == [ + "comment", + "like", + ] + + assert response["results"][0]["reaction_counts"] == {"like": 1, "comment": 1} + + reactions = { + "recent": True, + "counts": True, + "kinds": ["", "reshare ", "comment\n"], + } + response = await async_client.get_activities( + foreign_id_times=[(fid, dt)], reactions=reactions + ) + assert len(response["results"]) == 1 + assert response["results"][0]["id"] == activity["id"] + assert sorted(response["results"][0]["latest_reactions"].keys()) == [ + "comment", + "reshare", + ] + assert response["results"][0]["reaction_counts"] == {"comment": 1, "reshare": 1} + + +@pytest.mark.asyncio +async def test_activity_partial_update(async_client): + now = datetime.utcnow() + feed = async_client.feed("user", uuid4()) + await feed.add_activity( + { + "actor": "barry", + "object": "09", + "verb": "tweet", + "time": now, + "foreign_id": "fid:123", + "product": {"name": "shoes", "price": 9.99, "color": "blue"}, + } + ) + response = await feed.get() + activity = response["results"][0] + + to_set = { + "product.name": "boots", + "product.price": 7.99, + "popularity": 1000, + "foo": {"bar": {"baz": "qux"}}, + } + to_unset = ["product.color"] + + # partial update by ID + await async_client.activity_partial_update( + id=activity["id"], set=to_set, unset=to_unset + ) + response = await feed.get() + updated = response["results"][0] + expected = activity + expected["product"] = {"name": "boots", "price": 7.99} + expected["popularity"] = 1000 + expected["foo"] = {"bar": {"baz": "qux"}} + assert updated == expected + + # partial update by foreign ID + time + to_set = {"foo.bar.baz": 42, "popularity": 9000} + to_unset = ["product.price"] + await async_client.activity_partial_update( + foreign_id=activity["foreign_id"], + time=activity["time"], + set=to_set, + unset=to_unset, + ) + response = await feed.get() + updated = response["results"][0] + expected["product"] = {"name": "boots"} + expected["foo"] = {"bar": {"baz": 42}} + expected["popularity"] = 9000 + assert updated == expected + + +@pytest.mark.asyncio +async def test_activities_partial_update(async_client): + feed = async_client.feed("user", uuid4()) + await feed.add_activities( + [ + { + "actor": "barry", + "object": "09", + "verb": "tweet", + "time": datetime.utcnow(), + "foreign_id": "fid:123", + "product": {"name": "shoes", "price": 9.99, "color": "blue"}, + }, + { + "actor": "jerry", + "object": "10", + "verb": "tweet", + "time": datetime.utcnow(), + "foreign_id": "fid:456", + "product": {"name": "shoes", "price": 9.99, "color": "blue"}, + }, + { + "actor": "tommy", + "object": "09", + "verb": "tweet", + "time": datetime.utcnow(), + "foreign_id": "fid:789", + "product": {"name": "shoes", "price": 9.99, "color": "blue"}, + }, + ] + ) + response = await feed.get() + activities = response["results"] + + batch = [ + { + "id": activities[0]["id"], + "set": {"product.color": "purple", "custom": {"some": "extra data"}}, + "unset": ["product.price"], + }, + { + "id": activities[2]["id"], + "set": {"product.price": 9001, "on_sale": True}, + }, + ] + + # partial update by ID + await async_client.activities_partial_update(batch) + response = await feed.get() + updated = response["results"] + expected = activities + expected[0]["product"] = {"name": "shoes", "color": "purple"} + expected[0]["custom"] = {"some": "extra data"} + expected[2]["product"] = {"name": "shoes", "price": 9001, "color": "blue"} + expected[2]["on_sale"] = True + assert updated == expected + + # partial update by foreign ID + time + batch = [ + { + "foreign_id": activities[1]["foreign_id"], + "time": activities[1]["time"], + "set": {"product.color": "beeeeeeige", "custom": {"modified_by": "me"}}, + "unset": ["product.name"], + }, + { + "foreign_id": activities[2]["foreign_id"], + "time": activities[2]["time"], + "unset": ["on_sale"], + }, + ] + await async_client.activities_partial_update(batch) + response = await feed.get() + updated = response["results"] + + expected[1]["product"] = {"price": 9.99, "color": "beeeeeeige"} + expected[1]["custom"] = {"modified_by": "me"} + del expected[2]["on_sale"] + assert updated == expected + + +@pytest.mark.asyncio +async def test_reaction_add(async_client): + await async_client.reactions.add( + "like", "54a60c1e-4ee3-494b-a1e3-50c06acb5ed4", "mike" + ) + + +@pytest.mark.asyncio +async def test_reaction_add_to_target_feeds(async_client): + r = await async_client.reactions.add( + "superlike", + "54a60c1e-4ee3-494b-a1e3-50c06acb5ed4", + "mike", + data={"popularity": 50}, + target_feeds=["user:michelle"], + target_feeds_extra_data={"popularity": 100}, + ) + assert r["data"]["popularity"] == 50 + response = await async_client.feed("user", "michelle").get(limit=1) + a = response["results"][0] + assert r["id"] in a["reaction"] + assert a["verb"] == "superlike" + assert a["popularity"] == 100 + + child = await async_client.reactions.add_child( + "superlike", + r["id"], + "rob", + data={"popularity": 60}, + target_feeds=["user:michelle"], + target_feeds_extra_data={"popularity": 200}, + ) + + assert child["data"]["popularity"] == 60 + response = await async_client.feed("user", "michelle").get(limit=1) + a = response["results"][0] + assert child["id"] in a["reaction"] + assert a["verb"] == "superlike" + assert a["popularity"] == 200 + + +@pytest.mark.asyncio +async def test_reaction_get(async_client): + response = await async_client.reactions.add( + "like", "54a60c1e-4ee3-494b-a1e3-50c06acb5ed4", "mike" + ) + reaction = await async_client.reactions.get(response["id"]) + assert reaction["parent"] == "" + assert reaction["data"] == {} + assert reaction["latest_children"] == {} + assert reaction["children_counts"] == {} + assert reaction["activity_id"] == "54a60c1e-4ee3-494b-a1e3-50c06acb5ed4" + assert reaction["kind"] == "like" + assert "created_at" in reaction + assert "updated_at" in reaction + assert "id" in reaction + + +@pytest.mark.asyncio +async def test_reaction_update(async_client): + response = await async_client.reactions.add( + "like", "54a60c1e-4ee3-494b-a1e3-50c06acb5ed4", "mike" + ) + await async_client.reactions.update(response["id"], {"changed": True}) + + +@pytest.mark.asyncio +async def test_reaction_delete(async_client): + response = await async_client.reactions.add( + "like", "54a60c1e-4ee3-494b-a1e3-50c06acb5ed4", "mike" + ) + await async_client.reactions.delete(response["id"]) + + +@pytest.mark.asyncio +async def test_reaction_add_child(async_client): + response = await async_client.reactions.add( + "like", "54a60c1e-4ee3-494b-a1e3-50c06acb5ed4", "mike" + ) + await async_client.reactions.add_child("like", response["id"], "rob") + + +@pytest.mark.asyncio +async def test_reaction_filter_random(async_client): + await async_client.reactions.filter( + kind="like", + reaction_id="87a9eec0-fd5f-11e8-8080-80013fed2f5b", + id_lte="87a9eec0-fd5f-11e8-8080-80013fed2f5b", + ) + await async_client.reactions.filter( + activity_id="87a9eec0-fd5f-11e8-8080-80013fed2f5b", + id_lte="87a9eec0-fd5f-11e8-8080-80013fed2f5b", + ) + await async_client.reactions.filter( + user_id="mike", id_lte="87a9eec0-fd5f-11e8-8080-80013fed2f5b" + ) + + +def _first_result_should_be(response, element): + el = element.copy() + el.pop("duration") + assert len(response["results"]) == 1 + assert response["results"][0] == el + + +@pytest.mark.asyncio +async def test_reaction_filter(async_client): + activity_id = str(uuid1()) + user = str(uuid1()) + + response = await async_client.reactions.add("like", activity_id, user) + child = await async_client.reactions.add_child("like", response["id"], user) + reaction = await async_client.reactions.get(response["id"]) + + response = await async_client.reactions.add("comment", activity_id, user) + reaction_comment = await async_client.reactions.get(response["id"]) + + r = await async_client.reactions.filter(reaction_id=reaction["id"]) + _first_result_should_be(r, child) + + r = await async_client.reactions.filter( + kind="like", activity_id=activity_id, id_lte=reaction["id"] + ) + _first_result_should_be(r, reaction) + + r = await async_client.reactions.filter( + kind="like", user_id=user, id_lte=reaction["id"] + ) + _first_result_should_be(r, reaction) + + r = await async_client.reactions.filter(kind="comment", activity_id=activity_id) + _first_result_should_be(r, reaction_comment) + + +@pytest.mark.asyncio +async def test_user_add(async_client): + await async_client.users.add(str(uuid1())) + + +@pytest.mark.asyncio +async def test_user_add_get_or_create(async_client): + user_id = str(uuid1()) + r1 = await async_client.users.add(user_id) + r2 = await async_client.users.add(user_id, get_or_create=True) + assert r1["id"] == r2["id"] + assert r1["created_at"] == r2["created_at"] + assert r1["updated_at"] == r2["updated_at"] + + +@pytest.mark.asyncio +async def test_user_get(async_client): + response = await async_client.users.add(str(uuid1())) + user = await async_client.users.get(response["id"]) + assert user["data"] == {} + assert "created_at" in user + assert "updated_at" in user + assert "id" in user + + +@pytest.mark.asyncio +async def test_user_get_with_follow_counts(async_client): + response = await async_client.users.add(str(uuid1())) + user = await async_client.users.get(response["id"], with_follow_counts=True) + assert user["id"] == response["id"] + assert "followers_count" in user + assert "following_count" in user + + +@pytest.mark.asyncio +async def test_user_update(async_client): + response = await async_client.users.add(str(uuid1())) + await async_client.users.update(response["id"], {"changed": True}) + + +@pytest.mark.asyncio +async def test_user_delete(async_client): + response = await async_client.users.add(str(uuid1())) + await async_client.users.delete(response["id"]) + + +@pytest.mark.asyncio +async def test_collections_add(async_client): + await async_client.collections.add( + "items", {"data": 1}, id=str(uuid1()), user_id="tom" + ) + + +@pytest.mark.asyncio +async def test_collections_add_no_id(async_client): + await async_client.collections.add("items", {"data": 1}) + + +@pytest.mark.asyncio +async def test_collections_get(async_client): + response = await async_client.collections.add("items", {"data": 1}, id=str(uuid1())) + entry = await async_client.collections.get("items", response["id"]) + assert entry["data"] == {"data": 1} + assert "created_at" in entry + assert "updated_at" in entry + assert "id" in entry + + +@pytest.mark.asyncio +async def test_collections_update(async_client): + response = await async_client.collections.add("items", {"data": 1}, str(uuid1())) + await async_client.collections.update( + "items", response["id"], data={"changed": True} + ) + entry = await async_client.collections.get("items", response["id"]) + assert entry["data"] == {"changed": True} + + +@pytest.mark.asyncio +async def test_collections_delete(async_client): + response = await async_client.collections.add("items", {"data": 1}, str(uuid1())) + await async_client.collections.delete("items", response["id"]) + + +@pytest.mark.asyncio +async def test_feed_enrichment_collection(async_client): + entry = await async_client.collections.add("items", {"name": "time machine"}) + entry.pop("duration") + f = async_client.feed("user", "mike") + activity_data = { + "actor": "mike", + "verb": "buy", + "object": async_client.collections.create_reference(entry=entry), + } + await f.add_activity(activity_data) + response = await f.get() + assert set(activity_data.items()).issubset(set(response["results"][0].items())) + enriched_response = await f.get(enrich=True) + assert enriched_response["results"][0]["object"] == entry + + +@pytest.mark.asyncio +async def test_feed_enrichment_user(async_client): + user = await async_client.users.add(str(uuid1()), {"name": "Mike"}) + user.pop("duration") + f = async_client.feed("user", "mike") + activity_data = { + "actor": async_client.users.create_reference(user), + "verb": "buy", + "object": "time machine", + } + await f.add_activity(activity_data) + response = await f.get() + assert set(activity_data.items()).issubset(set(response["results"][0].items())) + enriched_response = await f.get(enrich=True) + assert enriched_response["results"][0]["actor"] == user + + +@pytest.mark.asyncio +async def test_feed_enrichment_own_reaction(async_client): + f = async_client.feed("user", "mike") + activity_data = {"actor": "mike", "verb": "buy", "object": "object"} + response = await f.add_activity(activity_data) + reaction = await async_client.reactions.add("like", response["id"], "mike") + reaction.pop("duration") + enriched_response = await f.get(reactions={"own": True}, user_id="mike") + assert enriched_response["results"][0]["own_reactions"]["like"][0] == reaction + + +@pytest.mark.asyncio +async def test_feed_enrichment_recent_reaction(async_client): + f = async_client.feed("user", "mike") + activity_data = {"actor": "mike", "verb": "buy", "object": "object"} + response = await f.add_activity(activity_data) + reaction = await async_client.reactions.add("like", response["id"], "mike") + reaction.pop("duration") + enriched_response = await f.get(reactions={"recent": True}) + assert enriched_response["results"][0]["latest_reactions"]["like"][0] == reaction + + +@pytest.mark.asyncio +async def test_feed_enrichment_reaction_counts(async_client): + f = async_client.feed("user", "mike") + activity_data = {"actor": "mike", "verb": "buy", "object": "object"} + response = await f.add_activity(activity_data) + reaction = await async_client.reactions.add("like", response["id"], "mike") + reaction.pop("duration") + enriched_response = await f.get(reactions={"counts": True}) + assert enriched_response["results"][0]["reaction_counts"]["like"] == 1 + + +@pytest.mark.asyncio +async def test_track_engagements(async_client): + engagements = [ + { + "content": "1", + "label": "click", + "features": [ + {"group": "topic", "value": "js"}, + {"group": "user", "value": "tommaso"}, + ], + "user_data": "tommaso", + }, + { + "content": "2", + "label": "click", + "features": [ + {"group": "topic", "value": "go"}, + {"group": "user", "value": "tommaso"}, + ], + "user_data": {"id": "486892", "alias": "Julian"}, + }, + { + "content": "3", + "label": "click", + "features": [{"group": "topic", "value": "go"}], + "user_data": {"id": "tommaso", "alias": "tommaso"}, + }, + ] + await async_client.track_engagements(engagements) + + +@pytest.mark.asyncio +async def test_track_impressions(async_client): + impressions = [ + { + "content_list": ["1", "2", "3"], + "features": [ + {"group": "topic", "value": "js"}, + {"group": "user", "value": "tommaso"}, + ], + "user_data": {"id": "tommaso", "alias": "tommaso"}, + }, + { + "content_list": ["2", "3", "5"], + "features": [{"group": "topic", "value": "js"}], + "user_data": {"id": "486892", "alias": "Julian"}, + }, + ] + await async_client.track_impressions(impressions) + + +@pytest.mark.asyncio +async def test_og(async_client): + response = await async_client.og("https://google.com") + assert "title" in response + assert "description" in response + + +@pytest.mark.asyncio +async def test_follow_stats(async_client): + uniq = uuid4() + f = async_client.feed("user", uniq) + await f.follow("user", uuid4()) + await f.follow("user", uuid4()) + await f.follow("user", uuid4()) + + await async_client.feed("user", uuid4()).follow("user", uniq) + await async_client.feed("timeline", uuid4()).follow("user", uniq) + + feed_id = "user:" + str(uniq) + response = await async_client.follow_stats(feed_id) + result = response["results"] + assert result["following"]["count"] == 3 + assert result["followers"]["count"] == 2 + + response = await async_client.follow_stats( + feed_id, followers_slugs=["timeline"], following_slugs=["timeline"] + ) + result = response["results"] + assert result["following"]["count"] == 0 + assert result["followers"]["count"] == 1 diff --git a/stream/tests/test_client.py b/stream/tests/test_client.py index a06af67..4321f61 100644 --- a/stream/tests/test_client.py +++ b/stream/tests/test_client.py @@ -31,10 +31,10 @@ from urllib.parse import urlparse, parse_qs -def connect_debug(): +def connect_debug(use_async=False): try: - key = os.environ["STREAM_KEY"] - secret = os.environ["STREAM_SECRET"] + key = "n8udba92h9hf" + secret = "5mjaw2a5bynt3fzdnxy2pwxh7fw89gg2dq3mm4jy56vhrwnadfst7yssatmaxhkv" except KeyError: print( "To run the tests the STREAM_KEY and STREAM_SECRET variables " @@ -45,7 +45,7 @@ def connect_debug(): ) sys.exit(1) - return stream.connect(key, secret, location="qa", timeout=30) + return stream.connect(key, secret, location="qa", timeout=30, use_async=use_async) client = connect_debug() @@ -703,8 +703,8 @@ def test_update_activity_to_targets(self): "object": 1, "foreign_id": foreign_id, "time": now, + "to": ["user:1", "user:2"], } - activity_data["to"] = ["user:1", "user:2"] self.user1.add_activity(activity_data) ret = self.user1.update_activity_to_targets( diff --git a/stream/users.py b/stream/users.py deleted file mode 100644 index 776e577..0000000 --- a/stream/users.py +++ /dev/null @@ -1,36 +0,0 @@ -class Users: - def __init__(self, client, token): - self.client = client - self.token = token - - def create_reference(self, id): - _id = id - if isinstance(id, (dict,)) and id.get("id") is not None: - _id = id.get("id") - return "SU:%s" % _id - - def add(self, user_id, data=None, get_or_create=False): - payload = dict(id=user_id, data=data) - return self.client.post( - "user/", - service_name="api", - signature=self.token, - data=payload, - params={"get_or_create": get_or_create}, - ) - - def get(self, user_id, **params): - return self.client.get( - "user/%s" % user_id, service_name="api", params=params, signature=self.token - ) - - def update(self, user_id, data=None): - payload = dict(data=data) - return self.client.put( - "user/%s" % user_id, service_name="api", signature=self.token, data=payload - ) - - def delete(self, user_id): - return self.client.delete( - "user/%s" % user_id, service_name="api", signature=self.token - ) diff --git a/stream/users/__init__.py b/stream/users/__init__.py new file mode 100644 index 0000000..f1cfbef --- /dev/null +++ b/stream/users/__init__.py @@ -0,0 +1 @@ +from .user import AsyncUsers, Users diff --git a/stream/users/base.py b/stream/users/base.py new file mode 100644 index 0000000..b17dead --- /dev/null +++ b/stream/users/base.py @@ -0,0 +1,39 @@ +from abc import ABC, abstractmethod + + +class AbstractUsers(ABC): + @abstractmethod + def create_reference(self, id): + pass + + @abstractmethod + def add(self, user_id, data=None, get_or_create=False): + pass + + @abstractmethod + def get(self, user_id, **params): + pass + + @abstractmethod + def update(self, user_id, data=None): + pass + + @abstractmethod + def delete(self, user_id): + pass + + +class BaseUsers(AbstractUsers, ABC): + + API_ENDPOINT = "user/" + SERVICE_NAME = "api" + + def __init__(self, client, token): + self.client = client + self.token = token + + def create_reference(self, id): + _id = id + if isinstance(id, (dict,)) and id.get("id") is not None: + _id = id.get("id") + return f"SU:{_id}" diff --git a/stream/users/user.py b/stream/users/user.py new file mode 100644 index 0000000..b96048d --- /dev/null +++ b/stream/users/user.py @@ -0,0 +1,73 @@ +from stream.users.base import BaseUsers + + +class Users(BaseUsers): + def add(self, user_id, data=None, get_or_create=False): + payload = dict(id=user_id, data=data) + return self.client.post( + self.API_ENDPOINT, + service_name=self.SERVICE_NAME, + signature=self.token, + data=payload, + params={"get_or_create": get_or_create}, + ) + + def get(self, user_id, **params): + return self.client.get( + f"{self.API_ENDPOINT}/{user_id}", + service_name=self.SERVICE_NAME, + params=params, + signature=self.token, + ) + + def update(self, user_id, data=None): + payload = dict(data=data) + return self.client.put( + f"{self.API_ENDPOINT}/{user_id}", + service_name=self.SERVICE_NAME, + signature=self.token, + data=payload, + ) + + def delete(self, user_id): + return self.client.delete( + f"{self.API_ENDPOINT}/{user_id}", + service_name=self.SERVICE_NAME, + signature=self.token, + ) + + +class AsyncUsers(BaseUsers): + async def add(self, user_id, data=None, get_or_create=False): + payload = dict(id=user_id, data=data) + return await self.client.post( + self.API_ENDPOINT, + service_name=self.SERVICE_NAME, + signature=self.token, + data=payload, + params={"get_or_create": str(get_or_create)}, + ) + + async def get(self, user_id, **params): + return await self.client.get( + f"{self.API_ENDPOINT}/{user_id}", + service_name=self.SERVICE_NAME, + params=params, + signature=self.token, + ) + + async def update(self, user_id, data=None): + payload = dict(data=data) + return await self.client.put( + f"{self.API_ENDPOINT}/{user_id}", + service_name=self.SERVICE_NAME, + signature=self.token, + data=payload, + ) + + async def delete(self, user_id): + return await self.client.delete( + f"{self.API_ENDPOINT}/{user_id}", + service_name=self.SERVICE_NAME, + signature=self.token, + ) diff --git a/stream/utils.py b/stream/utils.py index ea50635..bd0dbc5 100644 --- a/stream/utils.py +++ b/stream/utils.py @@ -13,8 +13,11 @@ def validate_feed_id(feed_id): """ feed_id = str(feed_id) if len(feed_id.split(":")) != 2: - msg = "Invalid feed_id spec %s, please specify the feed_id as feed_slug:feed_id" - raise ValueError(msg % feed_id) + msg = ( + f"Invalid feed_id spec {feed_id}, " + f"please specify the feed_id as feed_slug:feed_id" + ) + raise ValueError(msg) feed_slug, user_id = feed_id.split(":") validate_feed_slug(feed_slug) @@ -28,8 +31,8 @@ def validate_feed_slug(feed_slug): """ feed_slug = str(feed_slug) if not valid_re.match(feed_slug): - msg = "Invalid feed slug %s, please only use letters, numbers and _" - raise ValueError(msg % feed_slug) + msg = f"Invalid feed slug {feed_slug}, please only use letters, numbers and _" + raise ValueError(msg) return feed_slug @@ -39,8 +42,8 @@ def validate_user_id(user_id): """ user_id = str(user_id) if not valid_re.match(user_id): - msg = "Invalid user id %s, please only use letters, numbers and _" - raise ValueError(msg % user_id) + msg = f"Invalid user id {user_id}, please only use letters, numbers and _" + raise ValueError(msg) return user_id From 8a04a3454dbd9ee9fa28a2367a067ffcd1b462c8 Mon Sep 17 00:00:00 2001 From: Roman Zavadski Date: Sat, 15 Jan 2022 18:59:24 +0300 Subject: [PATCH 2/3] update documentation --- README.md | 125 ++++++++++++++++++++++++++++++ stream/client/client.py | 4 +- stream/collections/collections.py | 4 +- 3 files changed, 127 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index a94f6ed..9a755fb 100644 --- a/README.md +++ b/README.md @@ -147,6 +147,131 @@ events = [impression, engagement] redirect_url = client.create_redirect_url('http://google.com/', 'user_id', events) ``` +### Async code usage +```python +import datetime +import stream +client = stream.connect('YOUR_API_KEY', 'API_KEY_SECRET', use_async=True) + + +# Create a new client specifying data center location +client = stream.connect('YOUR_API_KEY', 'API_KEY_SECRET', location='us-east', use_async=True) +# Find your API keys here https://getstream.io/dashboard/ + +# Create a feed object +user_feed_1 = client.feed('user', '1') + +# Get activities from 5 to 10 (slow pagination) +result = await user_feed_1.get(limit=5, offset=5) +# (Recommended & faster) Filter on an id less than the given UUID +result = await user_feed_1.get(limit=5, id_lt="e561de8f-00f1-11e4-b400-0cc47a024be0") + +# Create a new activity +activity_data = {'actor': 1, 'verb': 'tweet', 'object': 1, 'foreign_id': 'tweet:1'} +activity_response = await user_feed_1.add_activity(activity_data) +# Create a bit more complex activity +activity_data = {'actor': 1, 'verb': 'run', 'object': 1, 'foreign_id': 'run:1', + 'course': {'name': 'Golden Gate park', 'distance': 10}, + 'participants': ['Thierry', 'Tommaso'], + 'started_at': datetime.datetime.now() +} +await user_feed_1.add_activity(activity_data) + +# Remove an activity by its id +await user_feed_1.remove_activity("e561de8f-00f1-11e4-b400-0cc47a024be0") +# or by foreign id +await user_feed_1.remove_activity(foreign_id='tweet:1') + +# Follow another feed +await user_feed_1.follow('flat', '42') + +# Stop following another feed +await user_feed_1.unfollow('flat', '42') + +# List followers/following +following = await user_feed_1.following(offset=0, limit=2) +followers = await user_feed_1.followers(offset=0, limit=10) + +# Creates many follow relationships in one request +follows = [ + {'source': 'flat:1', 'target': 'user:1'}, + {'source': 'flat:1', 'target': 'user:2'}, + {'source': 'flat:1', 'target': 'user:3'} +] +await client.follow_many(follows) + +# Batch adding activities +activities = [ + {'actor': 1, 'verb': 'tweet', 'object': 1}, + {'actor': 2, 'verb': 'watch', 'object': 3} +] +await user_feed_1.add_activities(activities) + +# Add an activity and push it to other feeds too using the `to` field +activity = { + "actor":"1", + "verb":"like", + "object":"3", + "to":["user:44", "user:45"] +} +await user_feed_1.add_activity(activity) + +# Retrieve an activity by its ID +await client.get_activities(ids=[activity_id]) + +# Retrieve an activity by the combination of foreign_id and time +await client.get_activities(foreign_id_times=[ + (foreign_id, activity_time), +]) + +# Enrich while getting activities +await client.get_activities(ids=[activity_id], enrich=True, reactions={"counts": True}) + +# Update some parts of an activity with activity_partial_update +set = { + 'product.name': 'boots', + 'colors': { + 'red': '0xFF0000', + 'green': '0x00FF00' + } +} +unset = [ 'popularity', 'details.info' ] +# ...by ID +await client.activity_partial_update(id=activity_id, set=set, unset=unset) +# ...or by combination of foreign_id and time +await client.activity_partial_update(foreign_id=foreign_id, time=activity_time, set=set, unset=unset) + +# Generating user token for client side usage (JS client) +user_token = client.create_user_token("user-42") + +# Javascript client side feed initialization +# client = stream.connect(apiKey, userToken, appId); + +# Generate a redirect url for the Stream Analytics platform to track +# events/impressions on url clicks +impression = { + 'content_list': ['tweet:1', 'tweet:2', 'tweet:3'], + 'user_data': 'tommaso', + 'location': 'email', + 'feed_id': 'user:global' +} + +engagement = { + 'content': 'tweet:2', + 'label': 'click', + 'position': 1, + 'user_data': 'tommaso', + 'location': 'email', + 'feed_id': + 'user:global' +} + +events = [impression, engagement] + +redirect_url = client.create_redirect_url('http://google.com/', 'user_id', events) + +``` + [JS client](http://github.com/getstream/stream-js). ### Contributing diff --git a/stream/client/client.py b/stream/client/client.py index e52534e..c63e059 100644 --- a/stream/client/client.py +++ b/stream/client/client.py @@ -269,9 +269,7 @@ def _make_request( params=default_params, timeout=self.timeout, ) - logger.debug( - f"stream api call {response.url}, headers {headers} data {data}" - ) + logger.debug(f"stream api call {response.url}, headers {headers} data {data}") return self._parse_response(response) def _parse_response(self, response): diff --git a/stream/collections/collections.py b/stream/collections/collections.py index 477062b..eebc730 100644 --- a/stream/collections/collections.py +++ b/stream/collections/collections.py @@ -19,9 +19,7 @@ def select(self, collection_name, ids): if not isinstance(ids, list): ids = [ids] - foreign_ids = ",".join( - f"{collection_name}:{k}" for i, k in enumerate(ids) - ) + foreign_ids = ",".join(f"{collection_name}:{k}" for i, k in enumerate(ids)) return self.client.get( self.URL, From 3d7ad183ded383bf1d988e2165bc697e15036ac1 Mon Sep 17 00:00:00 2001 From: Roman Zavadski Date: Sat, 22 Jan 2022 17:51:57 +0300 Subject: [PATCH 3/3] Some improvements after code review --- CHANGELOG.md | 5 ----- stream/__init__.py | 2 +- stream/client/async_client.py | 8 +++----- stream/client/client.py | 2 ++ stream/tests/conftest.py | 19 ++++++++++++++++--- stream/tests/test_async_client.py | 2 +- stream/tests/test_client.py | 8 ++++---- 7 files changed, 27 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 33e22be..97e250e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,8 +1,3 @@ -## 6.0.0 - 2021-01-19 - -* Add asynchronous code support -* Change ``%s`` to f-strings - ## 5.1.1 - 2022-01-18 * Handle backward compatible pyjwt 1.x support for token generation diff --git a/stream/__init__.py b/stream/__init__.py index 6261e94..4db3194 100644 --- a/stream/__init__.py +++ b/stream/__init__.py @@ -5,7 +5,7 @@ __copyright__ = "Copyright 2014, Stream.io, Inc" __credits__ = ["Thierry Schellenbach, mellowmorning.com, @tschellenbach"] __license__ = "BSD-3-Clause" -__version__ = "6.0.0" +__version__ = "5.1.1" __maintainer__ = "Thierry Schellenbach" __email__ = "support@getstream.io" __status__ = "Production" diff --git a/stream/client/async_client.py b/stream/client/async_client.py index 7fa02c5..3834314 100644 --- a/stream/client/async_client.py +++ b/stream/client/async_client.py @@ -41,8 +41,6 @@ def __init__( base_url=base_url, location=location, ) - self.session = aiohttp.ClientSession - token = self.create_jwt_token("collections", "*", feed_id="*", user_id="*") self.collections = AsyncCollections(self, token) @@ -234,7 +232,7 @@ async def _make_request( if method.lower() in ["post", "put", "delete"]: serialized = serializer.dumps(data) - async with self.session() as session: + async with aiohttp.ClientSession() as session: async with session.request( method, url, @@ -242,9 +240,9 @@ async def _make_request( headers=headers, params=default_params, timeout=self.timeout, - verify_ssl=False, ) as response: - + # remove JWT from logs + headers.pop("Authorization", None) logger.debug( f"stream api call {response}, headers {headers} data {data}", ) diff --git a/stream/client/client.py b/stream/client/client.py index c63e059..6326f30 100644 --- a/stream/client/client.py +++ b/stream/client/client.py @@ -269,6 +269,8 @@ def _make_request( params=default_params, timeout=self.timeout, ) + # remove JWT from logs + headers.pop("Authorization", None) logger.debug(f"stream api call {response.url}, headers {headers} data {data}") return self._parse_response(response) diff --git a/stream/tests/conftest.py b/stream/tests/conftest.py index af77ee4..88df1e2 100644 --- a/stream/tests/conftest.py +++ b/stream/tests/conftest.py @@ -1,6 +1,9 @@ import asyncio +import os +import sys import pytest + from stream import connect @@ -14,7 +17,7 @@ async def _parse_response(*args, **kwargs): @pytest.fixture(scope="module") -def event_loop(request): +def event_loop(): """Create an instance of the default event loop for each test case.""" loop = asyncio.get_event_loop_policy().new_event_loop() yield loop @@ -23,8 +26,18 @@ def event_loop(request): @pytest.fixture async def async_client(): - key = "n8udba92h9hf" - secret = "5mjaw2a5bynt3fzdnxy2pwxh7fw89gg2dq3mm4jy56vhrwnadfst7yssatmaxhkv" + key = os.getenv("STREAM_KEY") + secret = os.getenv("STREAM_SECRET") + if not key or not secret: + print( + "To run the tests the STREAM_KEY and STREAM_SECRET variables " + "need to be available. \n" + "Please create a pull request if you are an external " + "contributor, because these variables are automatically added " + "by Travis." + ) + sys.exit(1) + client = connect(key, secret, location="qa", timeout=30, use_async=True) wrapper(client._parse_response) yield client diff --git a/stream/tests/test_async_client.py b/stream/tests/test_async_client.py index 123b0f1..c1dfdf4 100644 --- a/stream/tests/test_async_client.py +++ b/stream/tests/test_async_client.py @@ -3,8 +3,8 @@ from datetime import datetime, timedelta from uuid import uuid1, uuid4 -import pytz import pytest +import pytz from dateutil.tz import tzlocal import stream diff --git a/stream/tests/test_client.py b/stream/tests/test_client.py index 4321f61..b15b51c 100644 --- a/stream/tests/test_client.py +++ b/stream/tests/test_client.py @@ -26,15 +26,15 @@ try: - from urlparse import urlparse, parse_qs + from urlparse import parse_qs, urlparse except ImportError: - from urllib.parse import urlparse, parse_qs + from urllib.parse import parse_qs, urlparse def connect_debug(use_async=False): try: - key = "n8udba92h9hf" - secret = "5mjaw2a5bynt3fzdnxy2pwxh7fw89gg2dq3mm4jy56vhrwnadfst7yssatmaxhkv" + key = os.environ["STREAM_KEY"] + secret = os.environ["STREAM_SECRET"] except KeyError: print( "To run the tests the STREAM_KEY and STREAM_SECRET variables "