Implement redis event storage (#100)

* Implement brute redis event storage

* Use redis sorted set

* Clean-up

* Fix

* Fix integration tests

* Update redis connection

* Refactoring

* Refactoring

* Remove max data checkers

* Refactoring

* flake8 happy

* Fix pytest config

Author: mrbish

Diff for: Dockerfile

@@ -11,6 +11,8 @@ ENV SUBSCRIPTIONS_PER_NEWSFEED="1024"
 ENV NEWSFEED_ID_LENGTH="128"
 ENV PROCESSOR_CONCURRENCY="4"
 
+ENV EVENT_STORAGE_DSN="NOTSET"
+
 ENV PYTHONUNBUFFERED=1
 ENV PYTHONPATH="${PYTHONPATH}:/code/src/"
 

Diff for: Makefile

@@ -1,6 +1,6 @@
 clean:
-	find . -type f -name "*.py[co]" -delete
-	find . -type d -name "__pycache__" -delete
+	find src -type f -name "*.py[co]" -delete
+	find src -type d -name "__pycache__" -delete
 
 test: clean
 	py.test tests/unit --cov=src/

Diff for: docker-compose.yml

@@ -7,9 +7,13 @@ services:
     environment:
       LOG_LEVEL: "DEBUG"
       API_BASE_PATH: "/api/"
+      EVENT_STORAGE_DSN: "redis://redis:6379?db=0&connection_timeout=5&minsize=1&maxsize=4"
     volumes:
       - "./:/code"
 
+  redis:
+    image: "redis:alpine"
+
   nginx:
     image: nginx
     ports:

Diff for: requirements.txt

@@ -1,5 +1,6 @@
 dependency-injector>=3.14
 uvloop>=0.14
 aiohttp[speedups]>=3.6
+aioredis>=1.3.0
 
 -r requirements-dev.txt

Diff for: scripts/integration_check.py

@@ -91,6 +91,7 @@ async def test(self):
             newsfeed_id=newsfeed_123,
             event_id=event_123_1['id'],
         )
+        await asyncio.sleep(0.1)
         assert event_123_1_deleted is True
 
         newsfeed_123_events = await self._api_client.get_events(newsfeed_id=newsfeed_123)
@@ -116,9 +117,11 @@ async def test(self):
             subscriber_124=subscriber_124,
             subscriber_125=subscriber_125,
         )
+        await asyncio.sleep(0.1)
 
         # Add event
         event_123_1 = await self._publish_event(newsfeed_id=newsfeed_123)
+        await asyncio.sleep(0.1)
 
         # Assert event publishing
         await self._assert_event_published_to_all_newsfeeds(
@@ -155,6 +158,7 @@ async def test(self):
             newsfeed_id=newsfeed_123,
             event=event_123_1,
         )
+        await asyncio.sleep(0.1)
 
         # Assert that all events are deleted
         await self._assert_no_events(newsfeed_123, subscriber_124, subscriber_125)

Diff for: setup.cfg

@@ -17,9 +17,13 @@ warn_redundant_casts = True
 warn_unused_ignores = True
 warn_return_any = True
 no_implicit_reexport = True
+ignore_missing_imports = True
 
 [mypy-newsfeed.containers]
 ignore_errors = True
 
 [mypy-newsfeed.core.loop]
 ignore_errors = True
+
+[tool:pytest]
+testpaths = tests

Diff for: src/newsfeed/configuration.py

@@ -18,6 +18,7 @@ def get_config() -> Dict[str, Any]:
             'event_storage': {
                 'max_newsfeeds': os.getenv('MAX_NEWSFEEDS'),
                 'max_events_per_newsfeed': os.getenv('EVENTS_PER_NEWSFEED'),
+                'dsn': os.getenv('EVENT_STORAGE_DSN'),
             },
             'subscription_storage': {
                 'max_newsfeeds': os.getenv('MAX_NEWSFEEDS'),

Diff for: src/newsfeed/containers.py

@@ -32,7 +32,7 @@ class Infrastructure(containers.DeclarativeContainer):
     )
 
     event_storage = providers.Singleton(
-        infrastructure.event_storages.InMemoryEventStorage,
+        infrastructure.event_storages.RedisEventStorage,
         config=config.event_storage,
     )
 

Diff for: src/newsfeed/infrastructure/event_storages.py

@@ -1,8 +1,15 @@
 """Infrastructure event storages module."""
 
+import json
+from contextlib import asynccontextmanager
 from collections import defaultdict, deque
 from typing import Dict, Deque, Iterable, Union
 
+import aioredis
+
+from .utils import parse_redis_dsn
+
+
 EventData = Dict[str, Union[str, int]]
 
 
@@ -34,7 +41,7 @@ class InMemoryEventStorage(EventStorage):
     """Event storage that stores events in memory."""
 
     def __init__(self, config: Dict[str, str]):
-        """Initialize queue."""
+        """Initialize storage."""
         super().__init__(config)
         self._storage: Dict[str, Deque[EventData]] = defaultdict(deque)
 
@@ -84,6 +91,84 @@ async def delete_by_fqid(self, newsfeed_id: str, event_id: str) -> None:
             del newsfeed_storage[event_index]
 
 
+class RedisEventStorage(EventStorage):
+    """Event storage that stores events in redis."""
+
+    def __init__(self, config: Dict[str, str]):
+        """Initialize storage."""
+        super().__init__(config)
+
+        redis_config = parse_redis_dsn(config['dsn'])
+        self._pool = aioredis.pool.ConnectionsPool(
+            address=redis_config['address'],
+            db=int(redis_config['db']),
+            create_connection_timeout=int(redis_config['connection_timeout']),
+            minsize=int(redis_config['minsize']),
+            maxsize=int(redis_config['maxsize']),
+            encoding='utf-8',
+        )
+
+    async def get_by_newsfeed_id(self, newsfeed_id: str) -> Iterable[EventData]:
+        """Get events data from storage."""
+        async with self._get_connection() as redis:
+            newsfeed_storage = []
+            # TODO:
+            #  - Check async
+            for event in await redis.lrange(key=f'newsfeed_id:{newsfeed_id}',
+                                            start=0,
+                                            stop=-1):
+                newsfeed_storage.append(json.loads(event))
+        return newsfeed_storage
+
+    async def get_by_fqid(self, newsfeed_id: str, event_id: str) -> EventData:
+        """Return data of specified event."""
+        async with self._get_connection() as redis:
+            event = await redis.get(f'event:{event_id}')
+
+        if not event:
+            raise EventNotFound(
+                newsfeed_id=newsfeed_id,
+                event_id=event_id,
+            )
+        else:
+            return json.loads(event)
+
+    async def add(self, event_data: EventData) -> None:
+        """Add event data to the storage."""
+        newsfeed_id = str(event_data['newsfeed_id'])
+
+        async with self._get_connection() as redis:
+            await redis.lpush(
+                key=f'newsfeed_id:{newsfeed_id}',
+                value=json.dumps(event_data),
+            )
+            await redis.append(
+                key=f"event:{event_data['id']}",
+                value=json.dumps(event_data),
+            )
+
+    async def delete_by_fqid(self, newsfeed_id: str, event_id: str) -> None:
+        """Delete data of specified event."""
+        async with self._get_connection() as redis:
+            event_key = f'event:{event_id}'
+            newsfeed = f'newsfeed_id:{newsfeed_id}'
+            event = await redis.get(event_key)
+            if not event:
+                raise EventNotFound(
+                    newsfeed_id=newsfeed_id,
+                    event_id=event_id,
+                )
+            else:
+                await redis.lrem(newsfeed, 1, event)
+                await redis.delete(event_key)
+
+    @asynccontextmanager
+    async def _get_connection(self) -> aioredis.commands.Redis:
+        async with self._pool.get() as connection:
+            redis = aioredis.commands.Redis(connection)
+            yield redis
+
+
 class EventStorageError(Exception):
     """Event-storage-related error."""
 

Diff for: src/newsfeed/infrastructure/utils.py

@@ -0,0 +1,14 @@
+"""Utils module for infrastructure."""
+
+from typing import Any, Dict
+from urllib.parse import urlparse, parse_qsl
+
+
+def parse_redis_dsn(dsn: str) -> Dict[str, Any]:
+    """Redis dsn parser."""
+    parsed_dsn = urlparse(dsn)
+    assert parsed_dsn.scheme == 'redis', ('Unsupported URI scheme', parsed_dsn.scheme)
+    return {
+        'address': (parsed_dsn.hostname, int(parsed_dsn.port)),
+        **dict(parse_qsl(parsed_dsn.query))
+    }