Merge pull request #17 from taskiq-python/feat/add-postgres-source

feat: add postgres storage source

Author: spikeninja

Dockerfile

@@ -9,6 +9,9 @@ RUN corepack enable
 
 WORKDIR /app
 
+ENV TASKIQ_ADMIN_DB_DRIVER=sqlite
+ENV TASKIQ_ADMIN_DB_FILE_PATH=/tmp/taskiq-admin.db
+
 # install dependencies
 COPY ./package.json /app/
 COPY ./pnpm-lock.yaml /app/
@@ -33,8 +36,9 @@ COPY --from=build /app/entrypoint.sh /usr/app/entrypoint.sh
 
 EXPOSE 3000
 ENV HOST=0.0.0.0 NODE_ENV=production
-ENV DB_FILE_PATH=/usr/database/database.db
-ENV BACKUP_FILE_PATH=/usr/database/backup.db
+ENV TASKIQ_ADMIN_DB_DRIVER=sqlite
+ENV TASKIQ_ADMIN_DB_FILE_PATH=/usr/database/database.db
+ENV TASKIQ_ADMIN_BACKUP_FILE_PATH=/usr/database/backup.db
 
 RUN npm install dotenv
 

README.md

@@ -1,6 +1,6 @@
 ## Broker-agnostic admin panel for Taskiq
 
-Standalone admin panel with all data stored in SQLite database
+Standalone admin panel with task results storage in SQLite or Postgres database
 
 
 - [Broker-agnostic admin panel for Taskiq](#broker-agnostic-admin-panel-for-taskiq)
@@ -91,6 +91,11 @@ volumes:
 2) run `make dev` to run it locally in dev mode
 3) run `make prod` to run it locally in prod mode
 
+Environment variables for DB setup:
+- `TASKIQ_ADMIN_DB_DRIVER`: required, one of `sqlite` or `postgres`
+- `TASKIQ_ADMIN_DB_FILE_PATH`: required when `TASKIQ_ADMIN_DB_DRIVER=sqlite`
+- `TASKIQ_ADMIN_DB_URL`: required when `TASKIQ_ADMIN_DB_DRIVER=postgres`
+
 ### 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

app/lib/utils.ts

@@ -2,7 +2,7 @@ import dayjs from 'dayjs'
 import utc from 'dayjs/plugin/utc.js'
 import { type ClassValue, clsx } from 'clsx'
 import { twMerge } from 'tailwind-merge'
-import type { TaskSelect } from '~/server/db/schema'
+import type { TaskSelect } from '../../shared/db/schema'
 
 dayjs.extend(utc)
 

drizzle.config.ts

@@ -1,11 +1,39 @@
 import 'dotenv/config'
 import { defineConfig } from 'drizzle-kit'
 
+const dbDriver = process.env.TASKIQ_ADMIN_DB_DRIVER
+
+if (!dbDriver) {
+  throw new Error('Environment variable TASKIQ_ADMIN_DB_DRIVER is required')
+}
+
+if (dbDriver !== 'sqlite' && dbDriver !== 'postgres') {
+  throw new Error(
+    'Environment variable TASKIQ_ADMIN_DB_DRIVER must be "sqlite" or "postgres"'
+  )
+}
+
+const dbUrl =
+  dbDriver === 'sqlite'
+    ? process.env.TASKIQ_ADMIN_DB_FILE_PATH
+    : process.env.TASKIQ_ADMIN_DB_URL
+
+if (!dbUrl) {
+  throw new Error(
+    dbDriver === 'sqlite'
+      ? 'Environment variable TASKIQ_ADMIN_DB_FILE_PATH is required for sqlite driver'
+      : 'Environment variable TASKIQ_ADMIN_DB_URL is required for postgres driver'
+  )
+}
+
 export default defineConfig({
   out: './drizzle',
-  schema: './shared/db/schema.ts',
-  dialect: 'sqlite',
+  schema:
+    dbDriver === 'sqlite'
+      ? './shared/db/schema.sqlite.ts'
+      : './shared/db/schema.postgres.ts',
+  dialect: dbDriver === 'sqlite' ? 'sqlite' : 'postgresql',
   dbCredentials: {
-    url: process.env.DB_FILE_PATH!
+    url: dbUrl
   }
 })

env-example

@@ -1,3 +1,5 @@
-DB_FILE_PATH=database/database.db
-BACKUP_FILE_PATH=database/backup.db
+TASKIQ_ADMIN_DB_DRIVER=sqlite
+TASKIQ_ADMIN_DB_FILE_PATH=database/database.db
+TASKIQ_ADMIN_DB_URL=postgres://postgres:postgres@localhost:5432/taskiq_admin
+TASKIQ_ADMIN_BACKUP_FILE_PATH=database/backup.db
 TASKIQ_ADMIN_API_TOKEN=supersecret

package.json

@@ -29,6 +29,7 @@
     "drizzle-orm": "^0.45.1",
     "lucide-vue-next": "^0.524.0",
     "nuxt": "^4.2.2",
+    "pg": "^8.16.3",
     "reka-ui": "^2.6.1",
     "tailwind-merge": "^3.4.0",
     "tailwindcss": "^4.1.17",

pnpm-lock.yaml

@@ -1,11 +1,27 @@
 import fs from 'fs'
-import { db } from '../../../shared/db'
+import { backupDatabase } from '../../../shared/db'
 import { utcNow } from '../../../shared/utils'
 import { envVariables } from '../../../shared/env'
-import { defineEventHandler, sendStream, setHeader } from '#imports'
+import {
+  createError,
+  defineEventHandler,
+  sendStream,
+  setHeader
+} from '#imports'
 
 export default defineEventHandler(async (event) => {
-  await db.$client.backup(envVariables.backupFilePath)
+  try {
+    await backupDatabase(envVariables.backupFilePath)
+  } catch (error) {
+    throw createError({
+      status: 400,
+      statusMessage:
+        error instanceof Error
+          ? error.message
+          : 'Could not create database backup'
+    })
+  }
+
   const stream = fs.createReadStream(envVariables.backupFilePath)
 
   const now = utcNow()

server/api/tasks/backup.get.ts

@@ -1,26 +1,22 @@
-import fs from 'fs/promises'
-import { db } from '../../shared/db'
+import { db, initializeDatabase } from '../../shared/db'
 import { defineNitroPlugin } from '#imports'
 import { taskiqAdminSettingsTable } from '~~/shared/db/schema'
 import { SETTINGS } from '~~/shared/constants/settings'
 
-export default defineNitroPlugin(async (nitroApp) => {
+export default defineNitroPlugin(async () => {
   console.log('Running DB initialization...')
-  const sqlScript = await fs.readFile('dbschema.sql', 'utf-8')
-
-  // executing the SQL script to create tables and indexes
-  db.$client.exec(sqlScript)
+  await initializeDatabase()
 
   // seeding default settings
-  db.insert(taskiqAdminSettingsTable)
+  await db
+    .insert(taskiqAdminSettingsTable)
     .values(
       Object.entries(SETTINGS).map(([_, value]) => ({
         key: value.key,
         value: value.defaultValue
       }))
     )
     .onConflictDoNothing()
-    .run()
 
   console.log('DB initialization completed.')
 })

server/plugins/startup.ts

@@ -0,0 +1,14 @@
+export const DB_TABLE_NAMES = {
+  tasks: 'taskiq_admin_tasks',
+  settings: 'taskiq_admin_settings'
+} as const
+
+export const DB_INDEX_NAMES = {
+  tasksState: 'idx_taskiq_admin_tasks__state',
+  tasksQueuedAt: 'idx_taskiq_admin_tasks__queued_at',
+  tasksStartedAt: 'idx_taskiq_admin_tasks__started_at',
+  tasksFinishedAt: 'idx_taskiq_admin_tasks__finished_at',
+  tasksExecutionTime: 'idx_taskiq_admin_tasks__execution_time',
+  tasksName: 'idx_taskiq_admin_tasks__name',
+  tasksWorker: 'idx_taskiq_admin_tasks__worker'
+} as const