39 lines
1.3 KiB
Python
39 lines
1.3 KiB
Python
import asyncio
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import List
|
|
|
|
from core.logger import get_logger
|
|
|
|
|
|
logger = get_logger("core.module_updates")
|
|
|
|
|
|
class ModuleUpdateManager:
|
|
def __init__(self, path: Path, remote: str = "origin", branch: str = "main") -> None:
|
|
self.path = path
|
|
self.remote = remote
|
|
self.branch = branch
|
|
|
|
async def check_updates(self) -> List[str]:
|
|
await self._run(["git", "fetch", self.remote])
|
|
log = await self._run(["git", "log", f"HEAD..{self.remote}/{self.branch}", "--oneline"])
|
|
return [line for line in log.splitlines() if line.strip()]
|
|
|
|
async def update_all(self) -> str:
|
|
return await self._run(["git", "pull", self.remote, self.branch])
|
|
|
|
async def rollback(self, ref: str = "HEAD~1") -> str:
|
|
return await self._run(["git", "reset", "--hard", ref])
|
|
|
|
async def _run(self, cmd: List[str]) -> str:
|
|
loop = asyncio.get_event_loop()
|
|
return await loop.run_in_executor(None, self._sync_run, cmd)
|
|
|
|
def _sync_run(self, cmd: List[str]) -> str:
|
|
logger.debug("Running command: %s", " ".join(cmd))
|
|
result = subprocess.run(cmd, cwd=self.path, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(result.stderr.strip() or "Command failed")
|
|
return result.stdout.strip()
|