feat(web-ui): #33 - added services selection + "update selected" and …

…"refresh selected" options

fix(server): 'image_tag' sometimes not returned correctly
chore(server): rewrote backend task creation and consumption
chore(build): bump react-query to 5.56.2

(#33) Batch Update Functionality:

* Either use middle-mouse click or the checkbox which appears on hover to select/deselected a service
* Selected cards/services are highlighted

.devcontainer/docker-compose.yaml

@@ -8,13 +8,12 @@ services:
     environment:
       - WEB_PORT=3000
       - SERVER_PORT=3001
-      # - AUTO_UPDATER_ENABLED=false
     volumes:
       - vsc-docking-station-vscode-server:/root/.vscode-server
       - ..:/workspaces/docking-station:cached
       - /var/run/docker.sock:/var/run/docker.sock
       - /etc/localtime:/etc/localtime:ro
-      - /mnt/big-vault/appdata/compose:/mnt/big-vault/appdata/compose
+      - /mnt/appdata/compose-files:/mnt/appdata/compose-files
     labels:
       # for homepage dashboard widget. More info: https://gethomepage.dev/latest/configs/docker/#automatic-service-discovery
       - homepage.group=Dev Containers

docker-compose.deploy.yaml

@@ -1,7 +1,7 @@
 
 services:
-  docking-station-builder:
+  docking-station:
     build:
       context: .
       dockerfile: Dockerfile.deploy
-    # image: docking-station
+    image: loolzzz/docking-station

docking-station-app/.eslintrc.json

@@ -1,6 +1,5 @@
 {
   "extends": [
-    "next/babel",
-    "next/core-web-vitals"
+    "next"
   ]
 }

docking-station-app/package.json

@@ -18,12 +18,12 @@
     "@mantine/modals": "^7.5.1",
     "@mantine/notifications": "^7.5.1",
     "@tabler/icons-react": "^2.47.0",
+    "@tanstack/react-query": "^5.56.2",
     "axios": "^1.6.7",
     "immer": "^10.1.1",
     "next": "14.1.0",
     "react": "^18",
     "react-dom": "^18",
-    "react-query": "^3.39.3",
     "react-use-websocket": "^4.8.1",
     "sass": "^1.79.2",
     "sharp": "^0.33.4",

docking-station-app/src/app/api/routes/stacks.py

@@ -2,16 +2,16 @@
 from logging import getLogger
 from threading import Thread
 
-from fastapi import APIRouter
-from fastapi.exceptions import HTTPException
+from fastapi import APIRouter, HTTPException
+from fastapi.responses import JSONResponse
 from fastapi_cache import FastAPICache
 
-from ..schemas import (DockerContainerResponse, DockerStackResponse,
-                       DockerStackUpdateRequest, MessageDict,
-                       MessageDictResponse,
-                       StartComposeStackServiceUpdateTaskResponse)
+from ..schemas import (DockerContainerResponse, DockerStackBatchUpdateRequest,
+                       DockerStackResponse, DockerStackUpdateRequest,
+                       MessageDict)
 from ..services import docker as docker_services
 from ..settings import cache_key_builder, cached, get_app_settings
+from ..task_store import StoreKey, TaskStore, TaskStoreItem
 
 __all__ = [
     'router',
@@ -20,7 +20,7 @@
 logger = getLogger(__name__)
 app_settings = get_app_settings()
 router = APIRouter()
-task_store: dict[tuple[str, str], tuple[Thread, asyncio.Queue[MessageDict]]] = {}
+task_store = TaskStore()
 
 
 @router.get('', response_model=list[DockerStackResponse])
@@ -63,52 +63,84 @@ async def get_compose_service_container(stack: str, service: str, no_cache: bool
         ) from exc
 
 
-@router.post('/{stack}/{service}/task', response_model=StartComposeStackServiceUpdateTaskResponse)
+@router.post('/{stack}/{service}/task')
 async def create_compose_stack_service_update_task(stack: str, service: str, request_body: DockerStackUpdateRequest):
-    task_thread, message_queue = docker_services.update_compose_stack_ws(
-        stack_name=stack,
-        service_name=service,
-        infer_envfile=request_body.infer_envfile,
-        restart_containers=request_body.restart_containers,
-        prune_images=request_body.prune_images,
+    return await create_compose_batch_update_task(
+        DockerStackBatchUpdateRequest(
+            services=[f'{stack}/{service}'],
+            **request_body.model_dump(by_alias=False),
+        )
     )
 
-    if is_new_task := (stack, service) not in task_store:
-        task_store[(stack, service)] = (task_thread, message_queue)
 
-    return StartComposeStackServiceUpdateTaskResponse(
-        task_id=f'{stack}/{service}',
-        created=is_new_task,
-    )
+@router.post('/batch_update')
+async def create_compose_batch_update_task(request_body: DockerStackBatchUpdateRequest):
 
+    def _acc_messages_task(task: TaskStoreItem,
+                           message_queue: asyncio.Queue[MessageDict],
+                           main_worker: Thread):
+        while True:
+            try:
+                task.append_message(
+                    message_queue.get_nowait()
+                )
+            except asyncio.QueueEmpty:
+                if not main_worker.is_alive() and message_queue.empty():
+                    break
+
+        main_worker.join()  # re-raise any exceptions from the main worker
+
+    for stack, services in request_body.stack_services.items():
+        skip = False
+        for service in services:
+            if (stack, service) in task_store:
+                skip = True
+                break
+        if skip:
+            continue
+
+        main_worker, queue = docker_services.update_compose_stack_ws(
+            stack_name=stack,
+            services=services,
+            infer_envfile=request_body.infer_envfile,
+            restart_containers=request_body.restart_containers,
+            prune_images=request_body.prune_images,
+        )
 
-@router.get('/{stack}/{service}/task', response_model=list[MessageDictResponse])
-async def poll_compose_stack_service_update_task(stack: str, service: str):
-    if (stack, service) not in task_store:
-        return []
+        task = TaskStoreItem()
+        acc_worker = Thread(target=_acc_messages_task, args=[task, queue, main_worker], daemon=True)
+        task.worker = acc_worker
+        acc_worker.start()
 
-    task_thread, message_queue = task_store[(stack, service)]
-    res: list[MessageDict] = []
+        for service in services:
+            task_store[(stack, service)] = task
+
+    return {}
+
+
+@router.get('/{stack}/{service}/task')
+async def poll_compose_stack_service_update_task(stack: str,
+                                                 service: str,
+                                                 offset: int | None = None):
+    key: StoreKey = (stack, service)
 
     try:
-        try:
-            while True:
-                res.append(
-                    message_queue.get_nowait()
-                )
+        if not (task := task_store.get(key, None)):
+            return JSONResponse(
+                content={'detail': f"Compose stack service task '{stack}/{service}' not found"},
+                status_code=404,
+            )
 
-        except asyncio.QueueEmpty:
-            if not task_thread.is_alive():
-                del task_store[(stack, service)]
-                cache_backend = FastAPICache.get_backend()
-                key, *_ = cache_key_builder(list_compose_stacks).split('(', 1)
-                await cache_backend.clear(namespace=key)
+        if not task.is_worker_alive():
+            cache_backend = FastAPICache.get_backend()
+            cache_key, *_ = cache_key_builder(list_compose_stacks).split('(', 1)
+            await cache_backend.clear(namespace=cache_key)
 
-                task_thread.join()  # re-raise any exceptions from the task
+            task.join()  # re-raise any exceptions from the task
 
     except Exception as _exc:
-        """ exception handling for 'task_thread.join()' """
+        """ exception handling for 'task.join()' """
         logger.exception("Error occurred while polling task thread for '%s/%s'", stack, service)
         raise
 
-    return res
+    return task.messages[offset:]

docking-station-app/src/app/api/schemas/images.py

@@ -35,7 +35,8 @@ def image_name(self) -> str:
     @computed_field
     @property
     def image_tag(self) -> str:
-        return self.repo_tag.split(':', 1)[1]
+        _, *tag = self.repo_tag.split(':', 1)
+        return tag[0] if tag else ''
 
 
 class DockerImageResponse(AliasedBaseModel):

docking-station-app/src/app/api/schemas/stacks.py

@@ -1,3 +1,4 @@
+from collections import defaultdict
 from pathlib import Path
 
 from pydantic import BaseModel, Field, computed_field, model_validator
@@ -7,11 +8,11 @@
 
 __all__ = [
     'DockerStack',
+    'DockerStackBatchUpdateRequest',
     'DockerStackResponse',
     'DockerStackRootModel',
     'DockerStackUpdateRequest',
     'DockerStackUpdateResponse',
-    'StartComposeStackServiceUpdateTaskResponse',
 ]
 
 
@@ -52,15 +53,25 @@ class DockerStackUpdateRequest(CamelCaseAliasedBaseModel):
     restart_containers: bool = True
 
 
+class DockerStackBatchUpdateRequest(CamelCaseAliasedBaseModel):
+    services: list[str]
+    infer_envfile: bool = True
+    prune_images: bool = False
+    restart_containers: bool = True
+
+    @property
+    def stack_services(self) -> dict[str, list[str]]:
+        res = defaultdict(list)
+        for item in self.services:
+            stack, service = item.split('/')
+            res[stack].append(service)
+        return dict(res)
+
+
 class DockerStackResponse(DockerStack):
     """Alias for `DockerStack`"""
 
 
 class DockerStackUpdateResponse(CamelCaseAliasedBaseModel):
     output: list[str]
     success: bool
-
-
-class StartComposeStackServiceUpdateTaskResponse(CamelCaseAliasedBaseModel):
-    task_id: str
-    created: bool

docking-station-app/src/app/api/services/docker.py

@@ -25,8 +25,8 @@
     'update_compose_stack',
 ]
 
-app_settings = get_app_settings()
 logger = getLogger(__name__)
+app_settings = get_app_settings()
 
 
 async def list_containers(filters: DockerContainerListFilters = None,
@@ -305,23 +305,20 @@ async def update_compose_stack(stack_name: str,
 
 
 def update_compose_stack_ws(stack_name: str,
-                            service_name: str = None,
+                            services: list[str] = [],
                             infer_envfile: bool = True,
                             restart_containers: bool = True,
                             prune_images: bool = False):
-    queue: asyncio.Queue[MessageDict] = asyncio.Queue()
 
-    async def _task():
-        nonlocal queue
+    async def _task(queue: asyncio.Queue[MessageDict]):
         nonlocal stack_name
-        nonlocal service_name
+        nonlocal services
         nonlocal infer_envfile
         nonlocal restart_containers
         nonlocal prune_images
 
         env_file = None
         config_files = None
-        output = []
 
         stack = next(iter(
             docker.compose.ls(
@@ -356,37 +353,56 @@ async def _task():
             *env_file_cmd,
             'up', '-d',
             *pull_cmd,
-            service_name,
+            *services,
         ])
         for line in stdout:
-            output.append(line)
             queue.put_nowait(
                 MessageDict(
                     stage='docker compose up --pull always',
                     message=line,
                 )
             )
 
-        if prune_images:
-            stdout = subprocess_stream_generator([
-                'docker', 'image', 'prune', '-f'
-            ])
-            for line in stdout:
-                output.append(line)
+        if app_settings.server.dryrun:
+            n = 50
+            for i in range(1, n + 1):
                 queue.put_nowait(
                     MessageDict(
-                        stage='docker image prune',
-                        message=line,
+                        stage='docker compose up --pull always',
+                        message=f'test line {i}/{n}',
                     )
                 )
+                await asyncio.sleep(0.1)
+
+        if prune_images:
+            if app_settings.server.dryrun:
+                n = 50
+                for i in range(1, n + 1):
+                    queue.put_nowait(
+                        MessageDict(
+                            stage='docker image prune',
+                            message=f'test line {i}/{n}',
+                        )
+                    )
+            else:
+                stdout = subprocess_stream_generator([
+                    'docker', 'image', 'prune', '-f'
+                ])
+                for line in stdout:
+                    queue.put_nowait(
+                        MessageDict(
+                            stage='docker image prune',
+                            message=line,
+                        )
+                    )
 
         await queue.put(
             MessageDict(
                 stage='Finished',
-                # message=output,
             )
         )
 
-    thread = Thread(target=lambda: asyncio.run(_task()), daemon=True)
-    thread.start()
-    return thread, queue
+    queue: asyncio.Queue[MessageDict] = asyncio.Queue()
+    worker = Thread(target=lambda: asyncio.run(_task(queue)), daemon=True)
+    worker.start()
+    return worker, queue