kin: auto-commit after pipeline

This commit is contained in:
Gros Frumos 2026-03-17 22:24:05 +02:00
parent fb4dd03342
commit 99262fb920
2 changed files with 400 additions and 0 deletions

View file

@ -8,9 +8,12 @@ Business logic for project deployments:
"""
import shlex
import shutil
import sqlite3
import subprocess
import time
from datetime import datetime
from pathlib import Path
VALID_RUNTIMES = {"docker", "node", "python", "static"}
@ -24,6 +27,44 @@ RUNTIME_STEPS = {
}
def pre_deploy_backup(project: dict) -> dict:
"""Create a timestamped backup of the project's SQLite database before deploy.
Looks for *.db / *.sqlite files in deploy_path or path.
Returns {"status": "ok", "path": "..."} on success,
{"status": "skipped"} if no DB found,
{"status": "error", "reason": "..."} on failure.
Raises OSError/PermissionError on unrecoverable errors (propagated to caller).
"""
project_path = Path(project.get("deploy_path") or project.get("path") or "")
if not project_path or not project_path.exists():
return {"status": "skipped", "reason": "project path not found"}
# Look for SQLite DB files (kin.db first, then any *.db / *.sqlite)
candidates = []
for pattern in ("kin.db", "*.db", "*.sqlite"):
candidates.extend(project_path.glob(pattern))
# Deduplicate while preserving order
seen: set[Path] = set()
db_files = []
for f in candidates:
if f not in seen:
seen.add(f)
db_files.append(f)
if not db_files:
return {"status": "skipped"}
db_file = db_files[0]
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path = db_file.parent / f"{db_file.name}.{timestamp}.bak"
shutil.copy2(str(db_file), str(backup_path))
return {"status": "ok", "path": str(backup_path)}
def build_deploy_steps(project: dict) -> list[str]:
"""Build deploy command list based on runtime and project config.
@ -84,6 +125,27 @@ def execute_deploy(project: dict, conn: sqlite3.Connection) -> dict:
"error": "No deploy steps: deploy_runtime not set or invalid",
}
# Mandatory pre-deploy backup — deploy without backup is impossible
try:
backup_result = pre_deploy_backup(project)
except Exception as e:
return {
"success": False,
"steps": steps,
"results": [],
"backup": {"status": "error", "reason": str(e)},
"error": f"Pre-deploy backup failed: {e}",
}
if backup_result.get("status") == "error":
return {
"success": False,
"steps": steps,
"results": [],
"backup": backup_result,
"error": f"Pre-deploy backup failed: {backup_result.get('reason')}",
}
deploy_path = project.get("deploy_path") or project.get("path") or None
results = []
overall_success = True
@ -142,6 +204,7 @@ def execute_deploy(project: dict, conn: sqlite3.Connection) -> dict:
"success": overall_success,
"steps": steps,
"results": results,
"backup": backup_result,
}