proof of concept UI, created using FastUI to display status of the jobs in the browser

.devcontainer/devcontainer.json

@@ -0,0 +1,55 @@
+{
+    "name": "CentralAPI",
+    "image": "python:3.11",
+    "workspaceFolder": "/app",
+    "remoteUser": "root",
+    "overrideCommand": true,
+    "postStartCommand": "make install && docker run -d -p 6379:6379 --name redis redis",
+    "appPort": ["8000:8000"],
+    "mounts": [
+      "source=${localWorkspaceFolder},target=/app,type=bind"
+      // "source=${localWorkspaceFolder}/.devcontainer/.zsh_history,target=/root/.zsh_history,type=bind"
+    ],
+    "features": {
+      "ghcr.io/devcontainers/features/common-utils:2": {
+        "configureZshAsDefaultShell": true
+      },
+      "ghcr.io/devcontainers-contrib/features/zsh-plugins:0": {
+        "omzPlugins": "https://github.com/zsh-users/zsh-autosuggestions https://github.com/zsh-users/zsh-syntax-highlighting https://github.com/zsh-users/zsh-completions https://github.com/zsh-users/zsh-history-substring-search",
+        "plugins": "zsh-autosuggestions zsh-syntax-highlighting zsh-completions zsh-history-substring-search"
+      },
+      "ghcr.io/devcontainers-contrib/features/flake8:2": {},
+      "ghcr.io/jungaretti/features/make":1,
+      "ghcr.io/devcontainers/features/docker-in-docker:2": {}
+    },
+    "customizations": {
+      // Configure properties specific to VS Code.
+      "vscode": {
+        // Set *default* container specific settings.json values on container create.
+        "settings": {
+          "python.defaultInterpreterPath": "/usr/local/bin/python",
+          "python.linting.enabled": true,
+          "python.formatting.yapfPath": "/usr/local/py-utils/bin/yapf",
+          "python.linting.flake8Path": "/usr/local/py-utils/bin/flake8",
+          "python.linting.pycodestylePath": "/usr/local/py-utils/bin/pycodestyle",
+          "python.linting.pydocstylePath": "/usr/local/py-utils/bin/pydocstyle"
+        },
+        // Add the IDs of extensions you want installed when the container is created.
+        "extensions": [
+            "ms-python.python",
+            "ms-python.isort",
+            "ms-python.flake8",
+            "ms-python.autopep8",
+            "ms-vscode.makefile-tools",
+            "mikestead.dotenv",
+            "PKief.material-icon-theme",
+            "aaron-bond.better-comments",
+            "redhat.vscode-yaml",
+            "eamodio.gitlens",
+            "oderwat.indent-rainbow",
+            "VisualStudioExptTeam.vscodeintellicode",
+            "njpwerner.autodocstring"
+        ]
+      }
+    }
+  }

arq/ui.py

@@ -0,0 +1,133 @@
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import HTMLResponse
+from fastui import FastUI, AnyComponent, prebuilt_html, components as c
+from fastui.components.display import DisplayLookup
+from fastui.events import GoToEvent, BackEvent
+from pydantic import BaseModel
+from arq import create_pool
+from arq.connections import RedisSettings
+from datetime import datetime
+from typing import Any
+
+app = FastAPI()
+
+
+class DisplayJobQueued(BaseModel):
+    function: str
+    args: tuple[Any, ...]
+    kwargs: dict[str, Any]
+    job_try: int
+    enqueue_time: datetime
+    score: int | None
+    job_id: str | None
+
+
+class DisplayJobResult(DisplayJobQueued):
+    success: bool
+    result: Any | None
+    start_time: datetime
+    finish_time: datetime
+    queue_name: str
+
+
+async def get_completed_jobs():
+    arq_jobs = await create_pool(RedisSettings())
+    jobs = await arq_jobs.all_job_results()
+    display_jobs = [
+        DisplayJobResult(
+            function=job.function,
+            args=job.args,
+            kwargs=job.kwargs,
+            # job_try=str(job.job_try), #if none then 0
+            job_try=0 if job.job_try is None else job.job_try,
+            enqueue_time=job.enqueue_time,
+            score=job.score,
+            job_id=job.job_id,
+            success=job.success,
+            result=str(job.result),
+            start_time=job.start_time,
+            finish_time=job.finish_time,
+            queue_name=job.queue_name
+            ) for job in jobs
+        ]
+    return display_jobs
+
+
+async def get_queued_jobs():
+    arq_jobs = await create_pool(RedisSettings())
+    jobs = await arq_jobs.queued_jobs()
+    display_jobs = [
+        DisplayJobQueued(
+            function=job.function,
+            args=job.args,
+            kwargs=job.kwargs,
+            job_try=0 if job.job_try is None else job.job_try,
+            # job_try=0,
+            enqueue_time=job.enqueue_time,
+            score=job.score,
+            job_id=job.job_id,
+            ) for job in jobs
+        ]
+    return display_jobs
+
+
+@app.get("/api/", response_model=FastUI, response_model_exclude_none=True)
+async def users_table() -> list[AnyComponent]:
+    jobs = await get_completed_jobs()
+    queued = await get_queued_jobs()
+    #TODO when there are no jobs, show a message that there are no jobs instead of 500 error
+    return [
+        c.Page(  
+            components=[
+                c.Heading(text='Arq Queued Jobs', level=2), 
+                c.Table(
+                    data=queued,
+                    columns=[
+                        DisplayLookup(field='job_id',  on_click=GoToEvent(url='/job/{job_id}/')),
+                        DisplayLookup(field='function'),
+                        DisplayLookup(field='args'),
+                        DisplayLookup(field='job_try'),
+                        DisplayLookup(field='queue_name'),
+                    ],
+                ),
+                c.Heading(text='Arq Complited Jobs', level=2),
+                c.Table(
+                    data=jobs,
+                    columns=[
+                        DisplayLookup(field='job_id',  on_click=GoToEvent(url='/job/{job_id}/')),
+                        DisplayLookup(field='function'),
+                        DisplayLookup(field='args'),
+                        DisplayLookup(field='success'),
+                        DisplayLookup(field='queue_name'),
+                    ],
+                ),
+            ]
+        ),
+    ]
+
+
+
+@app.get("/api/job/{job_id}/", response_model=FastUI, response_model_exclude_none=True)
+async def user_profile(job_id: str) -> list[AnyComponent]:
+    jobs = await get_completed_jobs()
+    queued = await get_queued_jobs()
+    jobs = jobs + queued
+    try:
+        job = next(job for job in jobs if job.job_id == job_id)
+    except StopIteration:
+        raise HTTPException(status_code=404, detail="User not found")
+    return [
+        c.Page(
+            components=[
+                c.Heading(text=job.job_id, level=2),
+                c.Link(components=[c.Text(text='Go back to the list')], on_click=BackEvent()),
+                c.Details(data=job),
+            ]
+        ),
+    ]
+
+
+@app.get('/{path:path}')
+async def html_landing() -> HTMLResponse:
+    """Simple HTML page which serves the React app, comes last as it matches all paths."""
+    return HTMLResponse(prebuilt_html(title='Arq Jobs'))

requirements/all.txt

@@ -2,3 +2,4 @@
 -r ./linting.txt
 -r ./testing.txt
 -r ./pyproject.txt
+-r ./ui.txt

requirements/ui.txt

@@ -0,0 +1,3 @@
+fastui==0.6.0
+fastapi==0.111.0
+uvicorn==0.29.0