Merge pull request #6 from taskiq-python/feat/add-queued-tasks

feat: add queued tasks

Author: spikeninja

README.md

@@ -6,7 +6,8 @@ Standalone admin panel with all data stored in SQLite database
 - [Broker-agnostic admin panel for Taskiq](#broker-agnostic-admin-panel-for-taskiq)
   - [Previews](#previews)
   - [Usage](#usage)
-  - [Docker Compose Examples](#docker-compose-examples)
+  - [Docker Compose Example](#docker-compose-example)
+  - [Task States](#task-states)
   - [Development](#development)
 
 ### Previews
@@ -16,7 +17,7 @@ Tasks Page | Task Details Page
 
 ### Usage
 
-1) Add this middleware to your taskiq broker:
+1) Add this middleware to your project:
 
 ```python
 from typing import Any
@@ -26,30 +27,49 @@ from datetime import datetime, UTC
 import httpx
 from taskiq import TaskiqMiddleware, TaskiqResult, TaskiqMessage
 
-TASKIQ_ADMIN_URL = "..." # or os.getenv() to use .env vars
-TASKIQ_ADMIN_API_TOKEN = "..." # or os.getenv() to use .env vars
-
-
 class TaskiqAdminMiddleware(TaskiqMiddleware):
-    def __init__(self, taskiq_broker_name: str | None = None):
+    def __init__(
+        self,
+        url: str,
+        api_token: str,
+        taskiq_broker_name: str | None = None,
+    ):
         super().__init__()
+        self.url = url
+        self.api_token = api_token
         self.__ta_broker_name = taskiq_broker_name
 
+    async def post_send(self, message):
+        now = datetime.now(UTC).replace(tzinfo=None).isoformat()
+        async with httpx.AsyncClient() as client:
+            await client.post(
+                headers={"access-token": self.api_token},
+                url=urljoin(self.url, f"/api/tasks/{message.task_id}/queued"),
+                json={
+                    "args": message.args,
+                    "kwargs": message.kwargs,
+                    "taskName": message.task_name,
+                    "worker": self.__ta_broker_name,
+                    "queuedAt": now,
+                },
+            )
+        return super().post_send(message)
+
     async def pre_execute(self, message: TaskiqMessage):
         """"""
+        now = datetime.now(UTC).replace(tzinfo=None).isoformat()
         async with httpx.AsyncClient() as client:
             await client.post(
-                headers={"access-token": TASKIQ_ADMIN_API_TOKEN},
-                url=urljoin(TASKIQ_ADMIN_URL, f"/api/tasks/{message.task_id}/started"),
+                headers={"access-token": self.api_token},
+                url=urljoin(self.url, f"/api/tasks/{message.task_id}/started"),
                 json={
+                    "startedAt": now,
                     "args": message.args,
                     "kwargs": message.kwargs,
                     "taskName": message.task_name,
                     "worker": self.__ta_broker_name,
-                    "startedAt": datetime.now(UTC).replace(tzinfo=None).isoformat(),
                 },
             )
-
         return super().pre_execute(message)
 
     async def post_execute(
@@ -58,26 +78,50 @@ class TaskiqAdminMiddleware(TaskiqMiddleware):
         result: TaskiqResult[Any],
     ):
         """"""
+        now = datetime.now(UTC).replace(tzinfo=None).isoformat()
         async with httpx.AsyncClient() as client:
             await client.post(
-                headers={"access-token": TASKIQ_ADMIN_API_TOKEN},
-                url=urljoin(TASKIQ_ADMIN_URL, f"/api/tasks/{message.task_id}/executed"),
+                headers={"access-token": self.api_token},
+                url=urljoin(
+                    self.url,
+                    f"/api/tasks/{message.task_id}/executed",
+                ),
                 json={
+                    "finishedAt": now,
                     "error": result.error
                     if result.error is None
                     else repr(result.error),
                     "executionTime": result.execution_time,
                     "returnValue": {"return_value": result.return_value},
-                    "finishedAt": datetime.now(UTC).replace(tzinfo=None).isoformat(),
                 },
             )
-
         return super().post_execute(message, result)
 ```
 
-2) Pull the image from GitHub Container Registry: `docker pull ghcr.io/taskiq-python/taskiq-admin:latest`
+2) Connect the middleware to your broker:
+  
+```python
+...
+broker = (
+    ListQueueBroker(
+        url=redis_url,
+        queue_name="my_lovely_queue",
+    )
+    .with_result_backend(result_backend)
+    .with_middlewares(
+        TaskiqAdminMiddleware(
+            url="http://localhost:3000", # the url to your taskiq-admin instance
+            api_token="supersecret", # any secret enough string
+            taskiq_broker_name="mybroker",
+        )
+    )
+)
+...
+```
 
-3) Replace `TASKIQ_ADMIN_API_TOKEN` with any secret enough string and run:
+3) Pull the image from GitHub Container Registry: `docker pull ghcr.io/taskiq-python/taskiq-admin:latest`
+
+4) Replace `TASKIQ_ADMIN_API_TOKEN` with any secret enough string and run:
 ```bash
 docker run -d --rm \
   -p "3000:3000" \
@@ -87,17 +131,10 @@ docker run -d --rm \
   "ghcr.io/taskiq-python/taskiq-admin:latest"
 ```
 
-4) Go to `http://localhost:3000/tasks`
+5) Go to `http://localhost:3000/tasks`
 
-### Docker Compose Examples
+### Docker Compose Example
 
-.env file example:
-```bash
-TASKIQ_ADMIN_URL="http://taskiq_admin:3000"
-TASKIQ_ADMIN_API_TOKEN="supersecret"
-```
-
-compose.yml file example
 ```yaml
 services:
   queue:
@@ -106,8 +143,9 @@ services:
       dockerfile: ./Dockerfile
     container_name: my_queue
     command: taskiq worker app.tasks.queue:broker --workers 1 --max-async-tasks 20
-    env_file:
-      - .env
+    environment:
+      - TASKIQ_ADMIN_URL=http://taskiq_admin:3000
+      - TASKIQ_ADMIN_API_TOKEN=supersecret
     depends_on:
       - redis
       - taskiq_admin
@@ -117,15 +155,23 @@ services:
     container_name: taskiq_admin
     ports:
       - 3000:3000
-    env_file:
-      - .env
+    environment:
+      - TASKIQ_ADMIN_API_TOKEN=supersecret
     volumes:
       - admin_data:/usr/database/
 
 volumes:
     admin_data:
 ```
 
+### Task States
+Let's assume we have a task 'do_smth', there are all states it can embrace:
+1) `queued` - the task has been sent to the queue without an error
+2) `running` - the task is grabbed by a worker and is being processed
+3) `success` - the task is fully processed without any errors
+4) `failure` - an error occured during the task processing
+5) `abandoned` - taskiq-admin sets all 'running' tasks as 'abandoned' if there was a downtime between the time these tasks were in 'running' state and the time of next startup of taskiq-admin
+
 ### Development
 1) Run `pnpm install` to install all dependencies
 2) Run `pnpm db:push` to create the sqlite database if needed

env-example

@@ -1,2 +1,3 @@
 DB_FILE_PATH=database/database.db
-BACKUP_FILE_PATH=database/backup.db
+BACKUP_FILE_PATH=database/backup.db
+TASKIQ_ADMIN_API_TOKEN=supersecret

nuxt.config.ts

@@ -1,17 +1,17 @@
-import tailwindcss from "@tailwindcss/vite"
+import tailwindcss from '@tailwindcss/vite'
 
 export default defineNuxtConfig({
-  compatibilityDate: "2024-11-01",
+  compatibilityDate: '2024-11-01',
   devtools: { enabled: true },
-  css: ["~/assets/css/main.css"],
-  srcDir: "src/",
+  css: ['~/assets/css/main.css'],
+  srcDir: 'src/',
   imports: {
-    scan: false,
+    autoImport: false
   },
   vite: {
-    plugins: [tailwindcss()],
+    plugins: [tailwindcss()]
   },
   typescript: {
-    strict: true,
-  },
+    strict: true
+  }
 })

package.json

@@ -2,29 +2,30 @@
   "name": "nuxt-app",
   "private": true,
   "type": "module",
-  "version": "1.5.0",
+  "version": "1.6.0",
   "scripts": {
     "build": "nuxt build",
     "dev": "nuxt dev",
     "generate": "nuxt generate",
     "preview": "nuxt preview",
     "postinstall": "nuxt prepare",
+    "typecheck": "tsc --noEmit",
     "db:push": "drizzle-kit push",
     "generate:sql": "drizzle-kit export --sql | sed 's/CREATE TABLE/CREATE TABLE IF NOT EXISTS/g; s/CREATE INDEX/CREATE INDEX IF NOT EXISTS/g' > dbschema.sql",
     "generate:future:sql": "drizzle-kit export --sql | sed 's/CREATE TABLE/CREATE TABLE IF NOT EXISTS/g; s/CREATE INDEX/CREATE INDEX IF NOT EXISTS/g' > dbschema.sql; sed -i '1s/^/PRAGMA journal_mode = WAL; PRAGMA synchronous = normal; PRAGMA journal_size_limit = 6144000;\\n/' dbschema.sql"
   },
   "dependencies": {
     "@internationalized/date": "^3.8.0",
     "@tailwindcss/vite": "^4.1.3",
-    "@tanstack/vue-table": "^8.21.2",
+    "@tanstack/vue-table": "^8.21.3",
     "@vueuse/core": "^12.8.2",
     "better-sqlite3": "^11.9.1",
     "bootstrap": "^5.3.3",
     "class-variance-authority": "^0.7.1",
     "clsx": "^2.1.1",
     "dayjs": "^1.11.13",
     "dotenv": "^16.4.7",
-    "drizzle-orm": "^0.41.0",
+    "drizzle-orm": "^0.42.0",
     "lucide-vue-next": "^0.487.0",
     "nuxt": "^3.16.2",
     "reka-ui": "^2.2.0",
@@ -41,7 +42,7 @@
     "@iconify-json/radix-icons": "^1.2.2",
     "@iconify/vue": "^4.3.0",
     "@types/better-sqlite3": "^7.6.12",
-    "drizzle-kit": "^0.30.6",
+    "drizzle-kit": "^0.31.0",
     "prettier": "^3.5.3",
     "typescript": "^5.8.3"
   }