dinlo
f624fee5e0
- SSH/SFTP управление без агентов - Пакеты (apt), Docker, файлы, systemd, ufw, cron - Плейбуки, групповые операции, планировщик - Бэкапы, отчёты, Portainer, Proxmox VE, Gitea Deploy - Терминал, Telegram-уведомления, тёмная/светлая тема
82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
"""Очередь длинных задач с журналом."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from datetime import datetime, timezone
|
|
from typing import Awaitable, Callable, Coroutine, List, Optional
|
|
|
|
from app.db.database import get_session
|
|
from app.db.models import TaskRecord
|
|
|
|
LogCallback = Callable[[str], None]
|
|
|
|
|
|
class TaskManager:
|
|
def __init__(self):
|
|
self._running: dict[int, asyncio.Task] = {}
|
|
|
|
def start(self, kind: str, title: str, server_id: Optional[int],
|
|
coro_factory: Callable[[LogCallback], Coroutine],
|
|
on_log: Optional[LogCallback] = None,
|
|
on_done: Optional[Callable[[TaskRecord], None]] = None) -> int:
|
|
"""Запустить задачу. coro_factory(log) — корутина, пишущая строки в log."""
|
|
with get_session() as s:
|
|
rec = TaskRecord(kind=kind, title=title, server_id=server_id, status="running")
|
|
s.add(rec)
|
|
s.commit()
|
|
s.refresh(rec)
|
|
task_id = rec.id
|
|
|
|
def log(line: str) -> None:
|
|
with get_session() as s:
|
|
r = s.get(TaskRecord, task_id)
|
|
if r:
|
|
r.log += line + "\n"
|
|
s.add(r)
|
|
s.commit()
|
|
if on_log:
|
|
on_log(line)
|
|
|
|
async def runner():
|
|
status = "success"
|
|
try:
|
|
await coro_factory(log)
|
|
except asyncio.CancelledError:
|
|
status = "cancelled"
|
|
log("[отменено пользователем]")
|
|
except Exception as e:
|
|
status = "error"
|
|
log(f"[ОШИБКА] {e}")
|
|
with get_session() as s:
|
|
r = s.get(TaskRecord, task_id)
|
|
if r:
|
|
r.status = status
|
|
r.finished_at = datetime.now(timezone.utc)
|
|
s.add(r)
|
|
s.commit()
|
|
s.refresh(r)
|
|
if on_done:
|
|
on_done(r)
|
|
|
|
self._running[task_id] = asyncio.ensure_future(runner())
|
|
return task_id
|
|
|
|
def cancel(self, task_id: int) -> None:
|
|
t = self._running.get(task_id)
|
|
if t and not t.done():
|
|
t.cancel()
|
|
|
|
def get(self, task_id: int) -> Optional[TaskRecord]:
|
|
with get_session() as s:
|
|
return s.get(TaskRecord, task_id)
|
|
|
|
def list_recent(self, limit: int = 50) -> List[TaskRecord]:
|
|
from sqlmodel import select
|
|
with get_session() as s:
|
|
return list(s.exec(
|
|
select(TaskRecord).order_by(TaskRecord.id.desc()).limit(limit)
|
|
).all())
|
|
|
|
|
|
tasks = TaskManager()
|