Multi-db session support and concurrent in-store changes for scaling the sessions to multiple users/windows

benchmarks/http/requirements.txt

@@ -1,10 +1,12 @@
-Cython==0.24
-bobo
-bottle
-cherrypy
-falcon
-flask
-muffin
-pyramid
-hug
-gunicorn
+Cython>=0.24,<0.26
+bobo==0.8.0
+bottle==0.14
+cherrypy==18.6.1
+falcon==3.1.0
+flask==2.1.3
+muffin==0.86.0
+pyramid==2.0
+hug==2.6.1
+gunicorn==20.1.0
+redis==4.5.1
+pymongo==4.3.3

hug/middleware.py

@@ -18,12 +18,13 @@
 
 """
 from __future__ import absolute_import
+from concurrent.futures import ThreadPoolExecutor
 
 import logging
 import re
 import uuid
 from datetime import datetime
-
+from hug.store import StoreWrapper
 
 class SessionMiddleware(object):
     """Simple session middleware.
@@ -50,11 +51,12 @@ class SessionMiddleware(object):
         "cookie_path",
         "cookie_secure",
         "cookie_http_only",
+        "session_data_executor"
     )
 
     def __init__(
         self,
-        store,
+        store_type="inmemory",
         context_name="session",
         cookie_name="sid",
         cookie_expires=None,
@@ -63,8 +65,9 @@ def __init__(
         cookie_path=None,
         cookie_secure=True,
         cookie_http_only=True,
+        max_workers = 10
     ):
-        self.store = store
+        self.store = StoreWrapper(store_type)
         self.context_name = context_name
         self.cookie_name = cookie_name
         self.cookie_expires = cookie_expires
@@ -73,10 +76,16 @@ def __init__(
         self.cookie_path = cookie_path
         self.cookie_secure = cookie_secure
         self.cookie_http_only = cookie_http_only
+        self.session_data_executor = ThreadPoolExecutor(max_workers=max_workers)
 
     def generate_sid(self):
         """Generate a UUID4 string."""
         return str(uuid.uuid4())
+
+    def get_session_data(self, sid):
+        if self.store.exists(sid):
+            return self.store.get(sid)
+        return {}
 
     def process_request(self, request, response):
         """Get session ID from cookie, load corresponding session data from coupled store and inject session data into
@@ -85,8 +94,9 @@ def process_request(self, request, response):
         sid = request.cookies.get(self.cookie_name, None)
         data = {}
         if sid is not None:
-            if self.store.exists(sid):
-                data = self.store.get(sid)
+            future = self.session_data_executor.submit(self.get_session_data, sid)
+            data = future.result()
+
         request.context.update({self.context_name: data})
 
     def process_response(self, request, response, resource, req_succeeded):
@@ -95,7 +105,9 @@ def process_response(self, request, response, resource, req_succeeded):
         if sid is None or not self.store.exists(sid):
             sid = self.generate_sid()
 
-        self.store.set(sid, request.context.get(self.context_name, {}))
+        # Session state might change for multiple users/windows, we will be able to update the store parallely
+        self.session_data_executor.submit(self.store.set, sid, request.context.get(self.context_name, {}))
+
         response.set_cookie(
             self.cookie_name,
             sid,

hug/store.py

@@ -1,55 +1,27 @@
-"""hug/store.py.
-
-A collecton of native stores which can be used with, among others, the session middleware.
-
-Copyright (C) 2016 Timothy Edmund Crosley
-
-Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
-documentation files (the "Software"), to deal in the Software without restriction, including without limitation
-the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and
-to permit persons to whom the Software is furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be included in all copies or
-substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
-TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
-THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF
-CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
-OTHER DEALINGS IN THE SOFTWARE.
-
-"""
-from hug.exceptions import StoreKeyNotFound
-
-
-class InMemoryStore:
-    """
-    Naive store class which can be used for the session middleware and unit tests.
-    It is not thread-safe and no data will survive the lifecycle of the hug process.
-    Regard this as a blueprint for more useful and probably more complex store implementations, for example stores
-    which make use of databases like Redis, PostgreSQL or others.
-    """
-
-    def __init__(self):
-        self._data = {}
-
-    def get(self, key):
-        """Get data for given store key. Raise hug.exceptions.StoreKeyNotFound if key does not exist."""
-        try:
-            data = self._data[key]
-        except KeyError:
-            raise StoreKeyNotFound(key)
-        return data
-
-    def exists(self, key):
-        """Return whether key exists or not."""
-        return key in self._data
-
-    def set(self, key, data):
-        """Set data object for given store key."""
-        self._data[key] = data
-
-    def delete(self, key):
-        """Delete data for given store key."""
-        if key in self._data:
-            del self._data[key]
+from hug.stores.inmemory_store import InMemoryStore
+from hug.stores.redis_store import RedisStore
+from hug.stores.mongo_store import MongoDBStore
+from hug.stores.sql_store import SQLStore
+
+class StoreWrapper:
+    def __init__(self, store_type='inmemory', **kwargs):
+        if store_type == 'redis':
+            self.store = RedisStore(**kwargs)
+        elif store_type == 'mongodb':
+            self.store = MongoDBStore(**kwargs)
+        elif store_type == 'sql':
+            self.store = SQLStore(**kwargs)
+        else:
+            self.store = InMemoryStore(**kwargs)
+
+    def get(self, key):
+        return self.store.get(key)
+
+    def set(self, key, data):
+        self.store.set(key, data)
+
+    def exists(self, key):
+        return self.store.exists(key)
+
+    def delete(self, key):
+        self.store.delete(key)

hug/stores/inmemory_store.py

@@ -0,0 +1,77 @@
+"""hug/store.py.
+
+A collecton of native stores which can be used with, among others, the session middleware.
+
+Copyright (C) 2016 Timothy Edmund Crosley
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+documentation files (the "Software"), to deal in the Software without restriction, including without limitation
+the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and
+to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or
+substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
+TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF
+CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+OTHER DEALINGS IN THE SOFTWARE.
+
+"""
+from hug.exceptions import StoreKeyNotFound
+from concurrent.futures import ThreadPoolExecutor
+import threading
+
+
+class InMemoryStore:
+    """
+    Naive store class which can be used for the session middleware and unit tests.
+    ~~It is not thread-safe and no data will survive the lifecycle of the hug process.~~
+    It is thread-safe and data will survive the lifecycle of the hug process.
+    Regard this as a blueprint for more useful and probably more complex store implementations, for example stores
+    which make use of databases like Redis, PostgreSQL or others.
+    """
+
+    def __init__(self, max_workers=10):
+        self._data = {}
+        self._lock = threading.Lock()
+        self._executor = ThreadPoolExecutor(max_workers=max_workers)
+
+    def get(self, key):
+        """Get data for given store key. Raise hug.exceptions.StoreKeyNotFound if key does not exist."""
+        with self._lock:
+            try:
+                data = self._data[key]
+            except KeyError:
+                raise StoreKeyNotFound(key)
+            return data
+
+    def exists(self, key):
+        with self._lock:
+            """Return whether key exists or not."""
+            return key in self._data
+
+    def set(self, key, data):
+        with self._lock:
+            """Set data object for given store key."""
+            self._data[key] = data
+
+    def delete(self, key):
+        with self._lock:
+            """Delete data for given store key."""
+            if key in self._data:
+                del self._data[key]
+
+    # Add async methods to be followed by milldleware to ensure parallelism
+    def async_get(self, key):
+        return self._executor.submit(self.get, key)
+
+    def async_set(self, key, data):
+        return self._executor.submit(self.set, key, data)
+
+    def async_delete(self, key):
+        return self._executor.submit(self.delete, key)
+
+    def async_exists(self, key):
+        return self._executor.submit(self.exists, key)

hug/stores/mongo_store.py

@@ -0,0 +1,40 @@
+from pymongo import MongoClient
+import logging
+import uuid
+
+class MongoDBStore:
+    def __init__(self, uri='mongodb://localhost:27017/', db_name='session_db', collection_name='sessions', ttl=3600, logger=None):
+        self._logger = logger if logger is not None else logging.getLogger("hug")
+        self._client = MongoClient(uri)
+        self._collection = self._client[db_name][collection_name]
+        self._collection.create_index("createdAt", expireAfterSeconds=ttl)
+
+    def get(self, key):
+        try:
+            return self._collection.find_one({"_id": key}) or {}
+        except Exception as e:
+            self._logger.exception("MongoDB exception: {}".format(str(e)))
+            return {}
+
+    def set(self, key, data):
+        try:
+            data['createdAt'] = uuid.uuid1().time
+            self._collection.update_one({"_id": key}, {"$set": data}, upsert=True)
+        except Exception as e:
+            self._logger.exception("MongoDB exception: {}".format(str(e)))
+            raise
+
+    def exists(self, key):
+        try:
+            return self._collection.count_documents({"_id": key}, limit=1) > 0
+        except Exception as e:
+            self._logger.exception("MongoDB exception: {}".format(str(e)))
+            raise
+
+
+    def delete(self, key):
+        try:
+            self._collection.delete_one({"_id": key})
+        except Exception as e:
+            self._logger.exception("MongoDB exception: {}".format(str(e)))
+            raise

hug/stores/redis_store.py

@@ -0,0 +1,50 @@
+import redis
+import logging
+
+class RedisStore:
+    def __init__(self, host='localhost', port=6379, db=0, ttl=3600, logger=None):
+        self._logger = logger if logger is not None else logging.getLogger("hug")
+        self._client = redis.StrictRedis(host=host, port=port, db=db, decode_responses=True)
+        self._ttl = ttl
+
+    def get(self, key):
+        try:
+            return self._client.hgetall(key)
+        except redis.RedisError as e:
+            self._logger.exception("Redis Error: {}".format(str(e)))
+            return {}
+        except Exception as e:
+            self._logger.exception("Redis Exception: {}".format(str(e)))
+            return {}
+
+    def set(self, key, data):
+        try:
+            self._client.hmset(key, data)
+            self._client.expire(key, self._ttl)
+        except redis.RedisError as e:
+            self._logger.exception("Redis Error: {}".format(str(e)))
+            raise
+        except Exception as e:
+            self._logger.exception("Redis Exception: {}".format(str(e)))
+            raise
+
+    def exists(self, key):
+        try:
+            return self._client.exists(key)
+        except redis.RedisError as e:
+            self._logger.exception("Redis Error: {}".format(str(e)))
+            raise
+        except Exception as e:
+            self._logger.exception("Redis Exception: {}".format(str(e)))
+            raise
+
+    def delete(self, key):
+        try:
+            self._client.delete(key)
+        except redis.RedisError as e:
+            self._logger.exception("Redis Error: {}".format(str(e)))
+            raise
+        except Exception as e:
+            self._logger.exception("Redis Exception: {}".format(str(e)))
+            raise
+