Spaces:
Paused
Paused
File size: 6,390 Bytes
2529305 | 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 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 | """
Script model for user-created Python scripts.
"""
import secrets
import uuid
from django.conf import settings
from django.db import models
from .environment import Environment
class Script(models.Model):
"""
Represents a Python script that can be executed.
Scripts are associated with an environment and can be run manually or on schedule.
"""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
name = models.CharField(max_length=200)
description = models.TextField(blank=True)
# The actual Python code
code = models.TextField(help_text="Python code to execute")
# Execution settings
environment = models.ForeignKey(
Environment,
on_delete=models.PROTECT,
related_name="scripts",
help_text="Python environment to use for execution",
)
# Tags for categorization
tags = models.ManyToManyField(
"Tag",
blank=True,
related_name="scripts",
help_text="Tags for organizing and filtering scripts",
)
timeout_seconds = models.PositiveIntegerField(
default=3600, # 1 hour default
help_text="Maximum execution time in seconds (default: 1 hour, max: 24 hours)",
)
# Status
is_enabled = models.BooleanField(
default=True,
help_text="Whether this script can be executed",
)
# Webhook
webhook_token = models.CharField(
max_length=64,
unique=True,
null=True,
blank=True,
db_index=True,
help_text="Unique token for webhook URL (auto-generated)",
)
# Notification settings
class NotifyOn(models.TextChoices):
NEVER = "never", "Never"
FAILURE = "failure", "On Failure"
SUCCESS = "success", "On Success"
BOTH = "both", "On Success and Failure"
notify_on = models.CharField(
max_length=20,
choices=NotifyOn.choices,
default=NotifyOn.NEVER,
help_text="When to send notifications for this script",
)
notify_email = models.EmailField(
blank=True,
help_text="Override email for this script (uses global default if empty)",
)
notify_webhook_url = models.URLField(
blank=True,
max_length=500,
help_text="URL to POST notification webhooks to",
)
notify_webhook_enabled = models.BooleanField(
default=False,
help_text="Enable webhook notifications for this script",
)
# Retention overrides (null = use global settings)
retention_days_override = models.PositiveIntegerField(
null=True,
blank=True,
help_text="Override global retention days for this script",
)
retention_count_override = models.PositiveIntegerField(
null=True,
blank=True,
help_text="Override global retention count for this script",
)
# Archive fields (soft delete)
archived_at = models.DateTimeField(
null=True,
blank=True,
help_text="When this script was archived (null = not archived)",
)
archived_by = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="archived_scripts",
)
# Metadata
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
created_by = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="scripts",
)
class Meta:
db_table = "scripts"
verbose_name = "script"
verbose_name_plural = "scripts"
ordering = ["-updated_at"]
def __str__(self):
if self.is_archived:
return f"{self.name} (archived)"
status = "enabled" if self.is_enabled else "disabled"
return f"{self.name} ({status})"
@property
def is_archived(self) -> bool:
"""Check if this script is archived."""
return self.archived_at is not None
@property
def can_run(self) -> bool:
"""Check if this script can be executed (enabled and not archived)."""
return self.is_enabled and not self.is_archived
@property
def last_run(self):
"""Return the most recent run for this script."""
return self.runs.order_by("-created_at").first()
@property
def last_successful_run(self):
"""Return the most recent successful run for this script."""
return self.runs.filter(status="success").order_by("-created_at").first()
@property
def run_count(self) -> int:
"""Return the total number of runs for this script."""
return self.runs.count()
@property
def success_rate(self) -> float | None:
"""Return the success rate as a percentage, or None if no runs."""
total = self.run_count
if total == 0:
return None
successful = self.runs.filter(status="success").count()
return (successful / total) * 100
def get_code_preview(self, max_lines: int = 5) -> str:
"""Return a preview of the script code (first N lines)."""
lines = self.code.split("\n")[:max_lines]
preview = "\n".join(lines)
if len(self.code.split("\n")) > max_lines:
preview += "\n..."
return preview
@staticmethod
def generate_webhook_token() -> str:
"""Generate a secure random webhook token (64 chars, URL-safe)."""
return secrets.token_urlsafe(48) # 48 bytes = 64 chars in base64
def create_webhook_token(self) -> str:
"""Create and save a new webhook token for this script."""
self.webhook_token = self.generate_webhook_token()
self.save(update_fields=["webhook_token", "updated_at"])
return self.webhook_token
def regenerate_webhook_token(self) -> str:
"""Regenerate the webhook token, invalidating the old one."""
return self.create_webhook_token()
def clear_webhook_token(self) -> None:
"""Remove the webhook token, disabling webhook access."""
self.webhook_token = None
self.save(update_fields=["webhook_token", "updated_at"])
@property
def has_webhook(self) -> bool:
"""Check if this script has a webhook token configured."""
return bool(self.webhook_token)
|