Introduce redis storage which syncs between multiple instances

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,13 @@
+#!/usr/bin/env python3
+from nicegui import app, ui
+
+
+@ui.page('/')
+async def index():
+    ui.input('general').bind_value(app.storage.general, 'text')
+    ui.input('user').bind_value(app.storage.user, 'text')
+    await ui.context.client.connected()
+    ui.input('tab').bind_value(app.storage.tab, 'text')
+
+
+ui.run(storage_secret='your private key to secure the browser session cookie')

nicegui/__init__.py

@@ -1,4 +1,4 @@
-from . import elements, html, run, ui
+from . import elements, html, run, storage, ui
 from .api_router import APIRouter
 from .app.app import App
 from .client import Client
@@ -20,5 +20,6 @@
     'elements',
     'html',
     'run',
+    'storage',
     'ui',
 ]

nicegui/app/app.py

@@ -46,6 +46,9 @@ def __init__(self, **kwargs) -> None:
         self._disconnect_handlers: List[Union[Callable[..., Any], Awaitable]] = []
         self._exception_handlers: List[Callable[..., Any]] = [log.exception]
 
+        self.on_startup(self.storage.general.initialize)
+        self.on_shutdown(self.storage.on_shutdown)
+
     @property
     def is_starting(self) -> bool:
         """Return whether NiceGUI is starting."""

nicegui/nicegui.py

@@ -176,6 +176,8 @@ async def _on_handshake(sid: str, data: Dict[str, Any]) -> bool:
         client.environ = sio.get_environ(sid)
         await sio.enter_room(sid, client.id)
     client.handle_handshake(data.get('next_message_id'))
+    assert client.tab_id is not None
+    await core.app.storage._create_tab_storage(client.tab_id)  # pylint: disable=protected-access
     return True
 
 

nicegui/optional_features.py

@@ -10,8 +10,9 @@
     'plotly',
     'polars',
     'pyecharts',
-    'webview',
+    'redis',
     'sass',
+    'webview',
 ]
 
 

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/file_persistent_dict.py

@@ -0,0 +1,48 @@
+from pathlib import Path
+from typing import Optional
+
+import aiofiles
+
+from nicegui import background_tasks, core, json
+from nicegui.logging import log
+
+from .persistent_dict import PersistentDict
+
+
+class FilePersistentDict(PersistentDict):
+
+    def __init__(self, filepath: Path, encoding: Optional[str] = None, *, indent: bool = False) -> None:
+        self.filepath = filepath
+        self.encoding = encoding
+        self.indent = indent
+        super().__init__(data={}, on_change=self.backup)
+
+    async def initialize(self) -> None:
+        try:
+            if self.filepath.exists():
+                async with aiofiles.open(self.filepath, encoding=self.encoding) as f:
+                    data = json.loads(await f.read())
+            else:
+                data = {}
+            self.update(data)
+        except Exception:
+            log.warning(f'Could not load storage file {self.filepath}')
+
+    def backup(self) -> None:
+        """Back up the data to the given file path."""
+        if not self.filepath.exists():
+            if not self:
+                return
+            self.filepath.parent.mkdir(exist_ok=True)
+
+        async def backup() -> None:
+            async with aiofiles.open(self.filepath, 'w', encoding=self.encoding) as f:
+                await f.write(json.dumps(self, indent=self.indent))
+        if core.loop:
+            background_tasks.create_lazy(backup(), name=self.filepath.stem)
+        else:
+            core.app.on_startup(backup())
+
+    def clear(self) -> None:
+        super().clear()
+        self.filepath.unlink(missing_ok=True)

nicegui/persistence/persistent_dict.py

@@ -0,0 +1,13 @@
+import abc
+
+from nicegui import observables
+
+
+class PersistentDict(observables.ObservableDict, abc.ABC):
+
+    @abc.abstractmethod
+    async def initialize(self) -> None:
+        """Load initial data from the persistence layer."""
+
+    async def close(self) -> None:
+        """Clean up the persistence layer."""

nicegui/persistence/read_only_dict.py

@@ -0,0 +1,24 @@
+from collections.abc import MutableMapping
+from typing import Any, Dict, Iterator
+
+
+class ReadOnlyDict(MutableMapping):
+
+    def __init__(self, data: Dict[Any, Any], write_error_message: str = 'Read-only dict') -> None:
+        self._data: Dict[Any, Any] = data
+        self._write_error_message: str = write_error_message
+
+    def __getitem__(self, item: Any) -> Any:
+        return self._data[item]
+
+    def __setitem__(self, key: Any, value: Any) -> None:
+        raise TypeError(self._write_error_message)
+
+    def __delitem__(self, key: Any) -> None:
+        raise TypeError(self._write_error_message)
+
+    def __iter__(self) -> Iterator:
+        return iter(self._data)
+
+    def __len__(self) -> int:
+        return len(self._data)

nicegui/persistence/redis_persistent_dict.py

@@ -0,0 +1,60 @@
+from .. import background_tasks, core, json, optional_features
+from ..logging import log
+from .persistent_dict import PersistentDict
+
+try:
+    import redis.asyncio as redis
+    optional_features.register('redis')
+except ImportError:
+    pass
+
+
+class RedisPersistentDict(PersistentDict):
+
+    def __init__(self, *, url: str, id: str, key_prefix: str = 'nicegui:') -> None:  # pylint: disable=redefined-builtin
+        if not optional_features.has('redis'):
+            raise ImportError('Redis is not installed. Please run "pip install nicegui[redis]".')
+        self.redis_client = redis.from_url(url)
+        self.pubsub = self.redis_client.pubsub()
+        self.key = key_prefix + id
+        super().__init__(data={}, 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)
+            self.update(json.loads(data) if data else {})
+        except Exception:
+            log.warning(f'Could not load data from Redis with key {self.key}')
+        await self.pubsub.subscribe(self.key + 'changes')
+
+        async def listen():
+            async for message in self.pubsub.listen():
+                if message['type'] == 'message':
+                    new_data = json.loads(message['data'])
+                    if new_data != self:
+                        self.update(new_data)
+
+        background_tasks.create(listen(), name=f'redis-listen-{self.key}')
+
+    def publish(self) -> None:
+        """Publish the data to Redis and notify other instances."""
+        async def backup() -> None:
+            pipeline = self.redis_client.pipeline()
+            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}')
+        else:
+            core.app.on_startup(backup())
+
+    async def close(self) -> None:
+        """Close Redis connection and subscription."""
+        await self.pubsub.unsubscribe()
+        await self.pubsub.close()
+        await self.redis_client.close()
+
+    def clear(self) -> None:
+        super().clear()
+        self.redis_client.delete(self.key)