implement redis storage for app.storage.user and began with redis_sto…

…rage example

docker-compose.yml

@@ -19,7 +19,6 @@ services:
     image: traefik:v2.3
     command:
       - --providers.docker
-      - --api.insecure=true
       - --accesslog # http access log
       - --log #Traefik log, for configurations and errors
       - --api # Enable the Dashboard and API

examples/redis_storage/Dockerfile

@@ -0,0 +1,3 @@
+FROM zauberzeug/nicegui:2.8.0
+
+RUN python -m pip install redis

examples/redis_storage/docker-compose.yml

@@ -0,0 +1,35 @@
+services:
+  x-nicegui: &nicegui-service
+    build:
+      context: .
+    environment:
+      - NICEGUI_REDIS_URL=redis://redis:6379
+    volumes:
+      - ./:/app
+      - ../../nicegui:/app/nicegui
+    labels:
+      - traefik.enable=true
+      - traefik.http.routers.nicegui.rule=PathPrefix(`/`)
+      - traefik.http.services.nicegui.loadbalancer.server.port=8080
+      - traefik.http.services.nicegui.loadbalancer.sticky.cookie=true
+
+  nicegui1:
+    <<: *nicegui-service
+
+  nicegui2:
+    <<: *nicegui-service
+
+  redis:
+    image: redis:alpine
+    ports:
+      - "6379:6379"
+
+  proxy:
+    image: traefik:v2.10
+    command:
+      - --providers.docker
+      - --entrypoints.web.address=:80
+    ports:
+      - "8080:80"
+    volumes:
+      - /var/run/docker.sock:/var/run/docker.sock

examples/redis_storage/main.py

@@ -0,0 +1,11 @@
+
+from nicegui import app, ui
+
+
+@ui.page('/')
+def index():
+    ui.input('general').bind_value(app.storage.general, 'text')
+    ui.input('user').bind_value(app.storage.user, 'text')
+
+
+ui.run(storage_secret='your private key to secure the browser session cookie')

nicegui/app/app.py

@@ -40,7 +40,7 @@ def __init__(self, **kwargs) -> None:
         self.config = AppConfig()
 
         self._startup_handlers: List[Union[Callable[..., Any], Awaitable]] = [self.storage.general.initialize,]
-        self._shutdown_handlers: List[Union[Callable[..., Any], Awaitable]] = [self.storage.general.close]
+        self._shutdown_handlers: List[Union[Callable[..., Any], Awaitable]] = [self.storage.on_shutdown]
         self._connect_handlers: List[Union[Callable[..., Any], Awaitable]] = []
         self._disconnect_handlers: List[Union[Callable[..., Any], Awaitable]] = []
         self._exception_handlers: List[Callable[..., Any]] = [log.exception]

nicegui/persistence/__init__.py

@@ -0,0 +1,11 @@
+from .file_persistent_dict import FilePersistentDict
+from .persistent_dict import PersistentDict
+from .read_only_dict import ReadOnlyDict
+from .redis_persistent_dict import RedisPersistentDict
+
+__all__ = [
+    'FilePersistentDict',
+    'PersistentDict',
+    'ReadOnlyDict',
+    'RedisPersistentDict',
+]

nicegui/persistence/redis_persistent_dict.py

@@ -8,21 +8,21 @@
 
 class RedisPersistentDict(PersistentDict):
 
-    def __init__(self, redis_url: str, key_prefix: str = 'nicegui:', encoding: str = 'utf-8') -> None:
+    def __init__(self, redis_url: str, id: str, key_prefix: str = 'nicegui:') -> None:  # pylint: disable=redefined-builtin
         self.redis_client = redis.from_url(redis_url)
         self.pubsub = self.redis_client.pubsub()
-        self.key_prefix = key_prefix
-        self.encoding = encoding
+        self.key = key_prefix + id
         super().__init__({}, on_change=self.publish)
 
     async def initialize(self) -> None:
         """Load initial data from Redis and start listening for changes."""
         try:
-            data = await self.redis_client.get(self.key_prefix + 'data')
+            data = await self.redis_client.get(self.key)
+            print(f'loading data: {data} for {self.key}')
             self.update(json.loads(data) if data else {})
         except Exception:
-            log.warning(f'Could not load data from Redis with prefix {self.key_prefix}')
-        await self.pubsub.subscribe(self.key_prefix + 'changes')
+            log.warning(f'Could not load data from Redis with key {self.key}')
+        await self.pubsub.subscribe(self.key + 'changes')
         async for message in self.pubsub.listen():
             if message['type'] == 'message':
                 new_data = json.loads(message['data'])
@@ -32,12 +32,13 @@ async def initialize(self) -> None:
     def publish(self) -> None:
         """Publish the data to Redis and notify other instances."""
         async def backup() -> None:
+            print(f'backup {self.key} with {json.dumps(self)}')
             pipeline = self.redis_client.pipeline()
-            pipeline.set(self.key_prefix + 'data', json.dumps(self))
-            pipeline.publish(self.key_prefix + 'changes', json.dumps(self))
+            pipeline.set(self.key, json.dumps(self))
+            pipeline.publish(self.key + 'changes', json.dumps(self))
             await pipeline.execute()
         if core.loop:
-            background_tasks.create_lazy(backup(), name=f'redis-{self.key_prefix}')
+            background_tasks.create_lazy(backup(), name=f'redis-{self.key}')
         else:
             core.app.on_startup(backup())
 

nicegui/storage.py

@@ -13,12 +13,12 @@
 from starlette.requests import Request
 from starlette.responses import Response
 
+from nicegui import background_tasks
+
 from . import core, observables
 from .context import context
 from .observables import ObservableDict
-from .persistence.file_persistent_dict import FilePersistentDict
-from .persistence.read_only_dict import ReadOnlyDict
-from .persistence.redis_persistent_dict import RedisPersistentDict
+from .persistence import FilePersistentDict, PersistentDict, ReadOnlyDict, RedisPersistentDict
 
 request_contextvar: contextvars.ContextVar[Optional[Request]] = contextvars.ContextVar('request_var', default=None)
 
@@ -50,19 +50,26 @@ def set_storage_secret(storage_secret: Optional[str] = None) -> None:
 
 class Storage:
     secret: Optional[str] = None
+    """Secret key for session storage."""
+    path = Path(os.environ.get('NICEGUI_STORAGE_PATH', '.nicegui')).resolve()
+    """Path to use for local persistence. Defaults to '.nicegui'."""
+    redis_url = os.environ.get('NICEGUI_REDIS_URL', None)
+    """URL to use for shared persistent storage via Redis. Defaults to None, which means local file storage is used."""
 
     def __init__(self) -> None:
-        self.path = Path(os.environ.get('NICEGUI_STORAGE_PATH', '.nicegui')).resolve()
-        """Path to use for local persistence. Defaults to '.nicegui'."""
         self.max_tab_storage_age: float = timedelta(days=30).total_seconds()
         """Maximum age in seconds before tab storage is automatically purged. Defaults to 30 days."""
-        self.redis_url = os.environ.get('NICEGUI_REDIS_URL', None)
-        """URL to use for shared persistent storage via Redis. Defaults to None, which means local file storage is used."""
-        self._general = RedisPersistentDict(self.redis_url) if self.redis_url \
-            else FilePersistentDict(self.path / 'storage-general.json', encoding='utf-8')
-        self._users: Dict[str, FilePersistentDict] = {}
+        self._general = Storage.create_persistent_dict('general')
+        self._users: Dict[str, PersistentDict] = {}
         self._tabs: Dict[str, observables.ObservableDict] = {}
 
+    @staticmethod
+    def create_persistent_dict(id: str) -> PersistentDict:
+        if Storage.redis_url:
+            return RedisPersistentDict(Storage.redis_url, f'user-{id}')
+        else:
+            return FilePersistentDict(Storage.path / f'storage-user-{id}.json', encoding='utf-8')
+
     @property
     def browser(self) -> Union[ReadOnlyDict, Dict]:
         """Small storage that is saved directly within the user's browser (encrypted cookie).
@@ -87,10 +94,10 @@ def browser(self) -> Union[ReadOnlyDict, Dict]:
         return request.session
 
     @property
-    def user(self) -> FilePersistentDict:
+    def user(self) -> PersistentDict:
         """Individual user storage that is persisted on the server (where NiceGUI is executed).
 
-        The data is stored in a file on the server.
+        The data is stored on the server.
         It is shared between all browser tabs by identifying the user via session cookie ID.
         """
         request: Optional[Request] = request_contextvar.get()
@@ -103,8 +110,8 @@ def user(self) -> FilePersistentDict:
             raise RuntimeError('app.storage.user can only be used within a UI context')
         session_id = request.session['id']
         if session_id not in self._users:
-            self._users[session_id] = FilePersistentDict(
-                self.path / f'storage-user-{session_id}.json', encoding='utf-8')
+            self._users[session_id] = Storage.create_persistent_dict(session_id)
+            background_tasks.create(self._users[session_id].initialize(), name=f'user-{session_id}')
         return self._users[session_id]
 
     @staticmethod
@@ -115,7 +122,7 @@ def _is_in_auto_index_context() -> bool:
             return False  # no client
 
     @property
-    def general(self) -> FilePersistentDict:
+    def general(self) -> PersistentDict:
         """General storage shared between all users that is persisted on the server (where NiceGUI is executed)."""
         return self._general
 
@@ -174,3 +181,9 @@ def clear(self) -> None:
             filepath.unlink()
         if self.path.exists():
             self.path.rmdir()
+
+    async def on_shutdown(self) -> None:
+        """Close all persistent storage."""
+        for user in self._users.values():
+            await user.close()
+        await self._general.close()