119 lines
4.6 KiB
Python
119 lines
4.6 KiB
Python
from __future__ import annotations
|
|
from django.db import models
|
|
from django.contrib.auth import get_user_model
|
|
from django.utils import timezone
|
|
|
|
# Remove eager evaluation to avoid accessing settings before configured
|
|
# User = get_user_model()
|
|
|
|
def get_user_model_cached():
|
|
return get_user_model()
|
|
|
|
class RemoteHost(models.Model):
|
|
AUTH_METHOD_CHOICES = [
|
|
('ssh_key', 'SSH Key'),
|
|
('agent', 'SSH Agent'),
|
|
('password', 'Password'), # not stored; prompt in future
|
|
]
|
|
name = models.CharField(max_length=100)
|
|
hostname = models.CharField(max_length=255)
|
|
port = models.PositiveIntegerField(default=22)
|
|
username = models.CharField(max_length=100)
|
|
auth_method = models.CharField(max_length=20, choices=AUTH_METHOD_CHOICES, default='ssh_key')
|
|
key_path = models.CharField(max_length=500, blank=True)
|
|
strict_host_key_checking = models.BooleanField(default=True)
|
|
notes = models.TextField(blank=True)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
|
|
class Meta:
|
|
ordering = ['name']
|
|
|
|
def __str__(self):
|
|
return f"{self.name} ({self.username}@{self.hostname})"
|
|
|
|
class CommandLog(models.Model):
|
|
STATUS_CHOICES = [
|
|
('running', 'Running'),
|
|
('ok', 'Success'),
|
|
('failed', 'Failed'),
|
|
('canceled', 'Canceled'),
|
|
('error', 'Error'),
|
|
]
|
|
RUN_TYPE_CHOICES = [
|
|
('single','Single'),
|
|
('batch','Batch'),
|
|
]
|
|
host = models.ForeignKey(RemoteHost, on_delete=models.CASCADE, related_name='command_logs')
|
|
created_by = models.ForeignKey('auth.User', null=True, blank=True, on_delete=models.SET_NULL)
|
|
command = models.TextField()
|
|
args_json = models.JSONField(null=True, blank=True)
|
|
run_type = models.CharField(max_length=10, choices=RUN_TYPE_CHOICES, default='single')
|
|
failed_step = models.IntegerField(null=True, blank=True)
|
|
started_at = models.DateTimeField(default=timezone.now)
|
|
finished_at = models.DateTimeField(null=True, blank=True)
|
|
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='running')
|
|
exit_code = models.IntegerField(null=True, blank=True)
|
|
output_tail = models.TextField(blank=True)
|
|
|
|
class Meta:
|
|
ordering = ['-started_at']
|
|
|
|
def mark_finished(self, status: str, exit_code: int | None, tail: str | None, failed_step: int | None = None):
|
|
self.status = status
|
|
self.exit_code = exit_code
|
|
self.failed_step = failed_step
|
|
self.finished_at = timezone.now()
|
|
if tail is not None:
|
|
self.output_tail = tail
|
|
self.save(update_fields=['status', 'exit_code', 'failed_step', 'finished_at', 'output_tail'])
|
|
|
|
def duration(self): # pragma: no cover simple helper
|
|
end = self.finished_at or timezone.now()
|
|
return (end - self.started_at).total_seconds()
|
|
|
|
class BatchScript(models.Model):
|
|
name = models.CharField(max_length=100, unique=True)
|
|
description = models.CharField(max_length=255, blank=True)
|
|
script = models.TextField(help_text="One command per line. Lines starting with # are ignored.")
|
|
created_by = models.ForeignKey('auth.User', null=True, blank=True, on_delete=models.SET_NULL)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
class Meta:
|
|
ordering = ['name']
|
|
|
|
def __str__(self): # pragma: no cover simple
|
|
return self.name
|
|
|
|
def clean(self): # normalize script formatting
|
|
if self.script:
|
|
normalized = self.script.replace('\r\n', '\n').replace('\r', '\n')
|
|
lines = [ln.rstrip() for ln in normalized.split('\n')]
|
|
# remove trailing empty lines
|
|
while lines and not lines[-1].strip():
|
|
lines.pop()
|
|
self.script = '\n'.join(lines).strip('\n')
|
|
|
|
def step_lines(self): # returns list of effective lines excluding comments/blank
|
|
if not self.script:
|
|
return []
|
|
return [l for l in self.script.split('\n') if l.strip() and not l.strip().startswith('#')]
|
|
|
|
def step_count(self): # pragma: no cover trivial
|
|
return len(self.step_lines())
|
|
|
|
class CommandTask(models.Model):
|
|
name = models.CharField(max_length=100, unique=True, help_text="Key used internally")
|
|
label = models.CharField(max_length=150)
|
|
command = models.TextField()
|
|
description = models.CharField(max_length=255, blank=True)
|
|
created_by = models.ForeignKey('auth.User', null=True, blank=True, on_delete=models.SET_NULL)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
class Meta:
|
|
ordering = ['name']
|
|
|
|
def __str__(self): # pragma: no cover
|
|
return self.label or self.name
|