File size: 5,030 Bytes
d6ee72a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 | """
Run model for tracking script execution history.
"""
import uuid
from django.conf import settings
from django.db import models
from .script import Script
class Run(models.Model):
"""
Represents a single execution of a script.
Tracks timing, output, and status of each run.
"""
class Status(models.TextChoices):
PENDING = "pending", "Pending"
RUNNING = "running", "Running"
SUCCESS = "success", "Success"
FAILED = "failed", "Failed"
TIMEOUT = "timeout", "Timeout"
CANCELLED = "cancelled", "Cancelled"
class TriggerType(models.TextChoices):
MANUAL = "manual", "Manual"
SCHEDULED = "scheduled", "Scheduled"
API = "api", "API"
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
script = models.ForeignKey(
Script,
on_delete=models.CASCADE,
related_name="runs",
)
# Execution status
status = models.CharField(
max_length=20,
choices=Status.choices,
default=Status.PENDING,
db_index=True,
)
exit_code = models.IntegerField(
null=True,
blank=True,
help_text="Process exit code (0 = success)",
)
# Output capture
stdout = models.TextField(
blank=True,
help_text="Standard output from script execution",
)
stderr = models.TextField(
blank=True,
help_text="Standard error from script execution",
)
# Timing
started_at = models.DateTimeField(
null=True,
blank=True,
help_text="When execution started",
)
ended_at = models.DateTimeField(
null=True,
blank=True,
help_text="When execution ended",
)
# Snapshot of script code at execution time (for audit trail)
code_snapshot = models.TextField(
blank=True,
help_text="Copy of script code at time of execution",
)
# Who triggered the run
triggered_by = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="triggered_runs",
)
# django-q2 task tracking
task_id = models.CharField(
max_length=100,
blank=True,
db_index=True,
help_text="django-q2 task ID for tracking async execution",
)
# How this run was triggered
trigger_type = models.CharField(
max_length=20,
choices=TriggerType.choices,
default=TriggerType.MANUAL,
help_text="How this run was triggered",
)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
db_table = "runs"
verbose_name = "run"
verbose_name_plural = "runs"
ordering = ["-created_at"]
indexes = [
models.Index(fields=["script", "-created_at"]),
models.Index(fields=["status", "-created_at"]),
]
def __str__(self):
return f"Run {self.id} - {self.script.name} ({self.status})"
@property
def duration(self) -> float | None:
"""Return the duration in seconds, or None if not completed."""
if self.started_at and self.ended_at:
return (self.ended_at - self.started_at).total_seconds()
return None
@property
def duration_display(self) -> str:
"""Return a human-readable duration string."""
d = self.duration
if d is None:
return "-"
if d < 60:
return f"{d:.1f}s"
minutes = int(d // 60)
seconds = d % 60
if minutes < 60:
return f"{minutes}m {seconds:.0f}s"
hours = minutes // 60
minutes = minutes % 60
return f"{hours}h {minutes}m"
@property
def is_finished(self) -> bool:
"""Check if the run has completed (successfully or not)."""
return self.status in [
self.Status.SUCCESS,
self.Status.FAILED,
self.Status.TIMEOUT,
self.Status.CANCELLED,
]
@property
def is_successful(self) -> bool:
"""Check if the run completed successfully."""
return self.status == self.Status.SUCCESS
@property
def has_output(self) -> bool:
"""Check if there is any output (stdout or stderr)."""
return bool(self.stdout or self.stderr)
def get_stdout_preview(self, max_lines: int = 10) -> str:
"""Return a preview of stdout (last N lines)."""
if not self.stdout:
return ""
lines = self.stdout.split("\n")
if len(lines) <= max_lines:
return self.stdout
return "...\n" + "\n".join(lines[-max_lines:])
def get_stderr_preview(self, max_lines: int = 10) -> str:
"""Return a preview of stderr (last N lines)."""
if not self.stderr:
return ""
lines = self.stderr.split("\n")
if len(lines) <= max_lines:
return self.stderr
return "...\n" + "\n".join(lines[-max_lines:])
|