82 lines
2.3 KiB
Python
82 lines
2.3 KiB
Python
"""Repositorio para gestionar job steps."""
|
|
|
|
from typing import Optional, List
|
|
from datetime import datetime
|
|
from dataclasses import dataclass
|
|
from .database import db
|
|
|
|
|
|
@dataclass
|
|
class JobStep:
|
|
"""Modelo de JobStep."""
|
|
id: int
|
|
job_id: str
|
|
step: str
|
|
started_at: str
|
|
finished_at: Optional[str]
|
|
duration_ms: Optional[int]
|
|
exit_code: Optional[int]
|
|
stdout: Optional[str]
|
|
stderr: Optional[str]
|
|
error: Optional[str]
|
|
|
|
|
|
class JobStepRepository:
|
|
"""Repositorio para operaciones con job steps."""
|
|
|
|
@staticmethod
|
|
def create(job_id: str, step: str) -> int:
|
|
"""
|
|
Crea un nuevo step para un job.
|
|
|
|
Args:
|
|
job_id: ID del job
|
|
step: Nombre del paso
|
|
|
|
Returns:
|
|
ID del step creado
|
|
"""
|
|
now = datetime.utcnow().isoformat()
|
|
cursor = db.execute("""
|
|
INSERT INTO job_steps (job_id, step, started_at)
|
|
VALUES (?, ?, ?)
|
|
""", (job_id, step, now))
|
|
|
|
return cursor.lastrowid
|
|
|
|
@staticmethod
|
|
def complete(
|
|
step_id: int,
|
|
exit_code: Optional[int] = None,
|
|
stdout: Optional[str] = None,
|
|
stderr: Optional[str] = None,
|
|
error: Optional[str] = None
|
|
):
|
|
"""Marca un step como completado."""
|
|
now = datetime.utcnow().isoformat()
|
|
|
|
# Obtener tiempo de inicio para calcular duración
|
|
row = db.fetchone("SELECT started_at FROM job_steps WHERE id = ?", (step_id,))
|
|
if row:
|
|
started = datetime.fromisoformat(row["started_at"])
|
|
finished = datetime.fromisoformat(now)
|
|
duration_ms = int((finished - started).total_seconds() * 1000)
|
|
else:
|
|
duration_ms = None
|
|
|
|
db.execute("""
|
|
UPDATE job_steps
|
|
SET finished_at = ?, duration_ms = ?, exit_code = ?,
|
|
stdout = ?, stderr = ?, error = ?
|
|
WHERE id = ?
|
|
""", (now, duration_ms, exit_code, stdout, stderr, error, step_id))
|
|
|
|
@staticmethod
|
|
def get_by_job(job_id: str) -> List[JobStep]:
|
|
"""Obtiene todos los steps de un job."""
|
|
rows = db.fetchall(
|
|
"SELECT * FROM job_steps WHERE job_id = ? ORDER BY id ASC",
|
|
(job_id,)
|
|
)
|
|
return [JobStep(**dict(row)) for row in rows]
|