From 88601451dcba4f81eca922a1d10fba3d12855ad3 Mon Sep 17 00:00:00 2001 From: JiriUhlir <149317995+JiriUhlir@users.noreply.github.com> Date: Thu, 28 May 2026 14:02:49 +0200 Subject: [PATCH] =?UTF-8?q?Implementov=C3=A1na=20port=C3=A1lov=C3=A1=20?= =?UTF-8?q?=C4=8D=C3=A1st=20SQLite=20job=20queue.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Přidán DB helper pro jobs a job_logs bez ORM. Přidána idempotentní migrace pro job tabulky včetně created_by_display_name. Přidány stránky /portal/jobs a /portal/jobs/{id}. Jobs page zobrazuje status, type, target, source, created_by, created_at, started_at, finished_at. Job detail zobrazuje payload, result/error a job_logs. Queued/running joby mají auto-refresh po 5 s. Navigace portálu obsahuje odkaz na Joby. App redeploy už nespouští deploy script přímo; vytváří queued job deploy_app s prázdným payloadem. Po redeploy se uživatel přesměruje na detail jobu. Přidán audit event app.redeploy.queued s metadata={"job_id": job_id}. Přidána ochrana proti duplicitnímu aktivnímu deploy jobu pro stejnou appku. --- app/db/jobs.py | 213 ++++++++++++++++++++++++++++++++++++++++ app/db/migrations.py | 57 +++++++++++ app/main.py | 3 +- app/routes/apps.py | 32 +++--- app/routes/jobs.py | 185 ++++++++++++++++++++++++++++++++++ app/templates/layout.py | 1 + 6 files changed, 474 insertions(+), 17 deletions(-) create mode 100644 app/db/jobs.py create mode 100644 app/db/migrations.py create mode 100644 app/routes/jobs.py diff --git a/app/db/jobs.py b/app/db/jobs.py new file mode 100644 index 0000000..3d18c18 --- /dev/null +++ b/app/db/jobs.py @@ -0,0 +1,213 @@ +import json +from typing import Any + +from app.db.database import get_connection +from app.db.migrations import run_migrations + + +ACTIVE_STATUSES = ("queued", "running") + + +def create_job( + job_type: str, + target_type: str, + target_id: str, + payload: dict[str, Any] | None = None, + user: dict[str, Any] | None = None, + source: str = "portal", +): + run_migrations() + payload_json = json.dumps(payload or {}, ensure_ascii=False, sort_keys=True) + created_by_user_id = user.get("id") if user else None + created_by_username = user.get("username") if user else None + created_by_display_name = user.get("display_name") if user else None + + con = get_connection() + cur = con.execute( + """ + INSERT INTO jobs ( + type, + target_type, + target_id, + payload_json, + status, + created_by_user_id, + created_by_username, + created_by_display_name, + source + ) + VALUES (?, ?, ?, ?, 'queued', ?, ?, ?, ?) + """, + ( + job_type, + target_type, + target_id, + payload_json, + created_by_user_id, + created_by_username, + created_by_display_name, + source, + ), + ) + job_id = cur.lastrowid + con.commit() + con.close() + append_job_log(job_id, "system", "Job queued.") + return job_id + + +def update_job_status( + job_id: int, + status: str, + worker_id: str | None = None, + result: dict[str, Any] | None = None, + error_text: str | None = None, +): + run_migrations() + result_json = json.dumps(result, ensure_ascii=False, sort_keys=True) if result is not None else None + + started_sql = ", started_at = COALESCE(started_at, CURRENT_TIMESTAMP)" if status == "running" else "" + finished_sql = ", finished_at = CURRENT_TIMESTAMP" if status in {"success", "failed", "cancelled"} else "" + + con = get_connection() + con.execute( + f""" + UPDATE jobs + SET status = ?, + worker_id = COALESCE(?, worker_id), + result_json = COALESCE(?, result_json), + error_text = COALESCE(?, error_text) + {started_sql} + {finished_sql} + WHERE id = ? + """, + (status, worker_id, result_json, error_text, job_id), + ) + con.commit() + con.close() + + +def append_job_log(job_id: int, stream: str, message: str): + run_migrations() + con = get_connection() + con.execute( + """ + INSERT INTO job_logs (job_id, stream, message) + VALUES (?, ?, ?) + """, + (job_id, stream, message), + ) + con.commit() + con.close() + + +def get_next_queued_job(): + run_migrations() + con = get_connection() + + row = con.execute( + """ + SELECT * + FROM jobs + WHERE status = 'queued' + AND NOT ( + type IN ('deploy_app', 'deploy_core_service') + AND EXISTS ( + SELECT 1 + FROM jobs active + WHERE active.id != jobs.id + AND active.type = jobs.type + AND active.target_type = jobs.target_type + AND active.target_id = jobs.target_id + AND active.status = 'running' + ) + ) + ORDER BY id + LIMIT 1 + """ + ).fetchone() + + con.close() + return dict(row) if row else None + + +def get_jobs(limit: int = 100): + run_migrations() + con = get_connection() + + rows = con.execute( + """ + SELECT * + FROM jobs + ORDER BY + CASE + WHEN status = 'running' THEN 0 + WHEN status = 'queued' THEN 1 + ELSE 2 + END, + id DESC + LIMIT ? + """, + (limit,), + ).fetchall() + + con.close() + return [dict(row) for row in rows] + + +def get_job(job_id: int): + run_migrations() + con = get_connection() + + row = con.execute( + """ + SELECT * + FROM jobs + WHERE id = ? + """, + (job_id,), + ).fetchone() + + con.close() + return dict(row) if row else None + + +def get_job_logs(job_id: int, limit: int = 1000): + run_migrations() + con = get_connection() + + rows = con.execute( + """ + SELECT * + FROM job_logs + WHERE job_id = ? + ORDER BY id + LIMIT ? + """, + (job_id, limit), + ).fetchall() + + con.close() + return [dict(row) for row in rows] + + +def has_active_deploy_job(target_type: str, target_id: str): + run_migrations() + con = get_connection() + + row = con.execute( + f""" + SELECT id + FROM jobs + WHERE type IN ('deploy_app', 'deploy_core_service') + AND target_type = ? + AND target_id = ? + AND status IN ({','.join('?' for _ in ACTIVE_STATUSES)}) + ORDER BY id DESC + LIMIT 1 + """, + (target_type, target_id, *ACTIVE_STATUSES), + ).fetchone() + + con.close() + return dict(row) if row else None diff --git a/app/db/migrations.py b/app/db/migrations.py new file mode 100644 index 0000000..e03fe81 --- /dev/null +++ b/app/db/migrations.py @@ -0,0 +1,57 @@ +from app.db.database import get_connection + + +def run_migrations(): + con = get_connection() + + con.execute( + """ + CREATE TABLE IF NOT EXISTS jobs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + type TEXT NOT NULL, + target_type TEXT NOT NULL, + target_id TEXT NOT NULL, + payload_json TEXT NOT NULL DEFAULT '{}', + status TEXT NOT NULL DEFAULT 'queued', + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + started_at TEXT, + finished_at TEXT, + created_by_user_id INTEGER, + created_by_username TEXT, + created_by_display_name TEXT, + source TEXT NOT NULL DEFAULT 'portal', + worker_id TEXT, + result_json TEXT, + error_text TEXT + ) + """ + ) + con.execute( + """ + CREATE TABLE IF NOT EXISTS job_logs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + job_id INTEGER NOT NULL, + stream TEXT NOT NULL, + message TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY(job_id) REFERENCES jobs(id) + ) + """ + ) + + con.execute("CREATE INDEX IF NOT EXISTS idx_jobs_status_created ON jobs(status, created_at)") + con.execute("CREATE INDEX IF NOT EXISTS idx_jobs_target_status ON jobs(target_type, target_id, status)") + con.execute("CREATE INDEX IF NOT EXISTS idx_job_logs_job_id ON job_logs(job_id, id)") + + try: + con.execute("ALTER TABLE jobs ADD COLUMN created_by_display_name TEXT") + except Exception: + pass + + try: + con.execute("ALTER TABLE deployments ADD COLUMN job_id INTEGER") + except Exception: + pass + + con.commit() + con.close() diff --git a/app/main.py b/app/main.py index 4ac1d1f..5ea3c09 100644 --- a/app/main.py +++ b/app/main.py @@ -5,7 +5,7 @@ from fastapi.staticfiles import StaticFiles from starlette.middleware.sessions import SessionMiddleware from .config import read_env_value -from .routes import apps, audit, auth, backups, deployments, health +from .routes import apps, audit, auth, backups, deployments, health, jobs def create_app() -> FastAPI: @@ -27,6 +27,7 @@ def create_app() -> FastAPI: app.include_router(apps.router) app.include_router(backups.router) app.include_router(deployments.router) + app.include_router(jobs.router) app.include_router(audit.router) return app diff --git a/app/routes/apps.py b/app/routes/apps.py index 387e62d..cdcb588 100644 --- a/app/routes/apps.py +++ b/app/routes/apps.py @@ -16,8 +16,9 @@ from ..config import ( ) from ..db.apps import get_app, get_app_deployments, get_apps, update_app_resources from ..db.audit import log_audit_event +from ..db.jobs import create_job, has_active_deploy_job from ..routes.deployments import render_status_pill -from ..shell import run_command, run_command_background +from ..shell import run_command from ..templates.layout import page, render_result router = APIRouter() @@ -250,28 +251,27 @@ def redeploy_app(app_id: str, user=Depends(require_user)): if not app: raise HTTPException(status_code=404, detail="App not found") - process = run_command_background( - [DEPLOY_SCRIPT, app_id], - extra_env={ - "APPFACTORY_TRIGGER_SOURCE": "portal", - "APPFACTORY_TRIGGERED_BY_USER_ID": user.get("id"), - "APPFACTORY_TRIGGERED_BY_USERNAME": user.get("username"), - "APPFACTORY_TRIGGERED_BY_DISPLAY_NAME": user.get("display_name") or user.get("username"), - }, + active_job = has_active_deploy_job("app", app_id) + if active_job: + return RedirectResponse(url=f"/portal/jobs/{active_job['id']}", status_code=303) + + job_id = create_job( + job_type="deploy_app", + target_type="app", + target_id=app_id, + payload={}, + user=user, + source="portal", ) log_audit_event( user, - action="app.redeploy", + action="app.redeploy.queued", target_type="app", target_id=app_id, - metadata={ - "app_id": app_id, - "pid": process.pid, - "trigger_source": "portal", - }, + metadata={"job_id": job_id}, ) - return RedirectResponse(url=f"/portal/apps/{quote(app_id, safe='')}", status_code=303) + return RedirectResponse(url=f"/portal/jobs/{job_id}", status_code=303) @router.get("/new-app", response_class=HTMLResponse) diff --git a/app/routes/jobs.py b/app/routes/jobs.py new file mode 100644 index 0000000..154250a --- /dev/null +++ b/app/routes/jobs.py @@ -0,0 +1,185 @@ +import html +import json +from urllib.parse import quote + +from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi.responses import HTMLResponse + +from app.auth import require_user +from app.db.jobs import get_job, get_job_logs, get_jobs +from app.routes.deployments import calculate_duration, render_status_pill +from app.templates.layout import page + +router = APIRouter() + +LIVE_STATUSES = {"queued", "running"} + + +def render_job_status(status: str | None) -> str: + return render_status_pill(status) + + +def pretty_json(value: str | None) -> str: + if not value: + return "{}" + + try: + return json.dumps(json.loads(value), ensure_ascii=False, indent=2, sort_keys=True) + except (TypeError, ValueError): + return value + + +@router.get("/jobs", response_class=HTMLResponse) +def jobs_page(request: Request, user=Depends(require_user)): + jobs = get_jobs() + refresh = "" if any( + (job.get("status") or "").lower() in LIVE_STATUSES for job in jobs + ) else "" + + rows = "" + for job in jobs: + job_id = html.escape(str(job.get("id", ""))) + status = render_job_status(job.get("status")) + source = html.escape(job.get("source", "") or "") + target_type = html.escape(job.get("target_type", "") or "") + target_id = html.escape(job.get("target_id", "") or "") + job_type = html.escape(job.get("type", "") or "") + created_by = html.escape(job.get("created_by_display_name") or job.get("created_by_username") or "") + created_at = html.escape(job.get("created_at", "") or "") + started_at = html.escape(job.get("started_at", "") or "") + finished_at = html.escape(job.get("finished_at", "") or "") + row_class = ' class="running-row"' if (job.get("status") or "").lower() == "running" else "" + + rows += f""" + + + #{job_id}
+ {job_type} + + {status} + {source} + {target_type}: {target_id} + {created_by} + {created_at} + {started_at} + {finished_at} + Detail + + """ + + if not rows: + rows = 'Zatím nejsou evidované žádné joby.' + + return page( + "Joby", + f""" + {refresh} +
+

Joby

+

Fronta portálových a webhook úloh připravená pro centrální worker.

+
+ +
+

Fronta

+ + + + + + + + + + + + + {rows} +
IDStatusSourceTargetCreated byCreated atStarted atFinished atAkce
+
+ """, + user=user, + ) + + +@router.get("/jobs/{job_id}", response_class=HTMLResponse) +def job_detail_page(job_id: int, request: Request, user=Depends(require_user)): + job = get_job(job_id) + if not job: + raise HTTPException(status_code=404, detail="Job not found") + + status_value = (job.get("status") or "").lower() + refresh = "" if status_value in LIVE_STATUSES else "" + title = f"Job #{html.escape(str(job.get('id', job_id)))}" + status = render_job_status(job.get("status")) + target_type = job.get("target_type", "") or "" + target_id = job.get("target_id", "") or "" + target_id_html = html.escape(target_id) + target_url = f"/portal/apps/{quote(target_id, safe='')}" if target_type == "app" else "/portal/jobs" + duration = html.escape(calculate_duration(job.get("started_at"), job.get("finished_at"))) + payload = html.escape(pretty_json(job.get("payload_json"))) + result = html.escape(pretty_json(job.get("result_json"))) + error_text = html.escape(job.get("error_text", "") or "") + + log_blocks = "" + for log in get_job_logs(job_id): + stream = html.escape(log.get("stream", "") or "system") + created_at = html.escape(log.get("created_at", "") or "") + message = html.escape(log.get("message", "") or "") + class_name = "log-stderr" if stream == "stderr" else "log-stdout" + log_blocks += f""" +
+
{created_at} · {stream}
+ 
{message}
+
+ """ + + if not log_blocks: + log_blocks = '

Zatím nejsou uložené žádné logy.

' + + return page( + title, + f""" + {refresh} +
+

{title}

+

+ ← Zpět na joby + Detail targetu +

+
+ +
+

Souhrn

+ + + + + + + + + + + +
Typ{html.escape(job.get("type", "") or "")}
Status{status}
Source{html.escape(job.get("source", "") or "")}
Target{html.escape(target_type)}: {target_id_html}
Created by{html.escape(job.get("created_by_display_name") or job.get("created_by_username") or "")}
Created at{html.escape(job.get("created_at", "") or "")}
Started at{html.escape(job.get("started_at", "") or "")}
Finished at{html.escape(job.get("finished_at", "") or "")}
Duration{duration}
Worker{html.escape(job.get("worker_id", "") or "")}
+
+ +
+
+

Payload

+ 
{payload}
+
+
+

Result / error

+ 
{result}
+ 
{error_text}
+
+
+ +
+

Logy

+ {log_blocks} +
+ """, + user=user, + ) diff --git a/app/templates/layout.py b/app/templates/layout.py index 36ec8a2..17447b9 100644 --- a/app/templates/layout.py +++ b/app/templates/layout.py @@ -15,6 +15,7 @@ def page(title: str, body: str, user=None) -> str: Aplikace Nová aplikace Nasazení + Joby Zálohy Audit