109 lines
4.3 KiB
Python
109 lines
4.3 KiB
Python
import uuid
|
|
|
|
from django.db import models
|
|
from django.db.models import Q
|
|
|
|
|
|
class Job(models.Model):
|
|
class Status(models.TextChoices):
|
|
QUEUED = "queued", "Queued"
|
|
RUNNING = "running", "Running"
|
|
SUCCEEDED = "succeeded", "Succeeded"
|
|
FAILED = "failed", "Failed"
|
|
|
|
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
|
type = models.CharField(max_length=120)
|
|
payload = models.JSONField(default=dict, blank=True)
|
|
status = models.CharField(max_length=20, choices=Status.choices, default=Status.QUEUED, db_index=True)
|
|
priority = models.SmallIntegerField(default=50)
|
|
available_at = models.DateTimeField(db_index=True)
|
|
attempts = models.PositiveIntegerField(default=0)
|
|
max_attempts = models.PositiveIntegerField(default=3)
|
|
idempotency_key = models.CharField(max_length=255, null=True, blank=True)
|
|
locked_by = models.CharField(max_length=255, null=True, blank=True)
|
|
locked_until = models.DateTimeField(null=True, blank=True)
|
|
last_error = models.TextField(blank=True, default="")
|
|
result = models.JSONField(null=True, blank=True)
|
|
created_at = models.DateTimeField(auto_now_add=True, db_index=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
finished_at = models.DateTimeField(null=True, blank=True)
|
|
|
|
class Meta:
|
|
ordering = ("-created_at",)
|
|
constraints = [
|
|
models.CheckConstraint(
|
|
name="job_status_valid_shape",
|
|
condition=(
|
|
Q(
|
|
status="queued",
|
|
locked_by__isnull=True,
|
|
locked_until__isnull=True,
|
|
finished_at__isnull=True,
|
|
)
|
|
| Q(
|
|
status="running",
|
|
locked_by__isnull=False,
|
|
locked_until__isnull=False,
|
|
finished_at__isnull=True,
|
|
)
|
|
| Q(
|
|
status__in=["succeeded", "failed"],
|
|
locked_by__isnull=True,
|
|
locked_until__isnull=True,
|
|
finished_at__isnull=False,
|
|
)
|
|
),
|
|
),
|
|
models.CheckConstraint(
|
|
name="job_attempts_valid",
|
|
condition=Q(attempts__gte=0, max_attempts__gte=1, attempts__lte=models.F("max_attempts")),
|
|
),
|
|
models.UniqueConstraint(
|
|
fields=["idempotency_key"],
|
|
condition=Q(idempotency_key__isnull=False),
|
|
name="job_idempotency_key_unique",
|
|
),
|
|
]
|
|
indexes = [
|
|
models.Index(
|
|
fields=["available_at", "-priority", "created_at", "id"],
|
|
name="job_claim_idx",
|
|
condition=Q(status="queued"),
|
|
),
|
|
models.Index(fields=["locked_until"], name="job_timeout_idx", condition=Q(status="running")),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"{self.type}:{self.id}"
|
|
|
|
|
|
class JobEvent(models.Model):
|
|
class Type(models.TextChoices):
|
|
CREATED = "created", "Created"
|
|
CLAIMED = "claimed", "Claimed"
|
|
PROGRESS = "progress", "Progress"
|
|
LEASE_RENEWED = "lease_renewed", "Lease renewed"
|
|
SUCCEEDED = "succeeded", "Succeeded"
|
|
RETRY_SCHEDULED = "retry_scheduled", "Retry scheduled"
|
|
FAILED = "failed", "Failed"
|
|
TIMEOUT_REQUEUED = "timeout_requeued", "Timeout requeued"
|
|
MANUAL_RETRY = "manual_retry", "Manual retry"
|
|
|
|
job = models.ForeignKey(Job, on_delete=models.CASCADE, related_name="events")
|
|
type = models.CharField(max_length=40, choices=Type.choices, db_index=True)
|
|
attempt = models.PositiveIntegerField(default=0)
|
|
worker_id = models.CharField(max_length=255, null=True, blank=True)
|
|
message = models.TextField(blank=True, default="")
|
|
data = models.JSONField(default=dict, blank=True)
|
|
created_at = models.DateTimeField(auto_now_add=True, db_index=True)
|
|
|
|
class Meta:
|
|
ordering = ("id",)
|
|
indexes = [
|
|
models.Index(fields=["job", "id"], name="job_event_job_id_idx"),
|
|
models.Index(fields=["id"], name="job_event_id_idx"),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"{self.type}:{self.job_id}:{self.id}"
|