95 lines
3.7 KiB
Python
95 lines
3.7 KiB
Python
from django.db import transaction
|
|
from django.utils import timezone
|
|
|
|
from apps.core.errors import ApiError
|
|
from apps.processing.models import StepExecution
|
|
|
|
|
|
class ProcessingService:
|
|
"""Step execution state machine. Transitions are validated server-side."""
|
|
|
|
_STARTABLE = {"pending", "ready", "rework_required"}
|
|
_PAUSABLE = {"in_progress"}
|
|
_RESUMABLE = {"paused"}
|
|
_COMPLETABLE = {"in_progress"}
|
|
_REJECTABLE = {"in_progress", "waiting"}
|
|
|
|
@transaction.atomic
|
|
def start(self, execution, user) -> StepExecution:
|
|
execution = self._locked(execution)
|
|
if execution.state not in self._STARTABLE:
|
|
raise ApiError("INVALID_STEP_TRANSITION", f"Cannot start from state '{execution.state}'.")
|
|
execution.state = StepExecution.State.IN_PROGRESS
|
|
execution.started_at = timezone.now()
|
|
execution.assigned_to = user
|
|
execution.save()
|
|
return execution
|
|
|
|
@transaction.atomic
|
|
def pause(self, execution) -> StepExecution:
|
|
execution = self._locked(execution)
|
|
if execution.state not in self._PAUSABLE:
|
|
raise ApiError("INVALID_STEP_TRANSITION", f"Cannot pause from state '{execution.state}'.")
|
|
execution.state = StepExecution.State.PAUSED
|
|
execution.save()
|
|
return execution
|
|
|
|
@transaction.atomic
|
|
def resume(self, execution) -> StepExecution:
|
|
execution = self._locked(execution)
|
|
if execution.state not in self._RESUMABLE:
|
|
raise ApiError("INVALID_STEP_TRANSITION", f"Cannot resume from state '{execution.state}'.")
|
|
execution.state = StepExecution.State.IN_PROGRESS
|
|
execution.save()
|
|
return execution
|
|
|
|
@transaction.atomic
|
|
def complete(self, execution, measurements=None, notes="") -> StepExecution:
|
|
execution = self._locked(execution)
|
|
if execution.state not in self._COMPLETABLE:
|
|
raise ApiError("INVALID_STEP_TRANSITION", f"Cannot complete from state '{execution.state}'.")
|
|
execution.state = StepExecution.State.COMPLETED
|
|
execution.completed_at = timezone.now()
|
|
if measurements is not None:
|
|
execution.measurements = measurements
|
|
if notes:
|
|
execution.notes = notes
|
|
execution.save()
|
|
return execution
|
|
|
|
@transaction.atomic
|
|
def reject(self, execution, notes="") -> StepExecution:
|
|
execution = self._locked(execution)
|
|
if execution.state not in self._REJECTABLE:
|
|
raise ApiError("INVALID_STEP_TRANSITION", f"Cannot reject from state '{execution.state}'.")
|
|
execution.state = StepExecution.State.REJECTED
|
|
execution.notes = notes or execution.notes
|
|
execution.save()
|
|
return execution
|
|
|
|
@transaction.atomic
|
|
def rework(self, execution, target_step) -> StepExecution:
|
|
"""Create a new execution on `target_step`; preserve history via parent link."""
|
|
execution = self._locked(execution)
|
|
if execution.state != StepExecution.State.COMPLETED:
|
|
raise ApiError("INVALID_STEP_TRANSITION", "Only a completed step can trigger rework.")
|
|
execution.state = StepExecution.State.REWORK_REQUIRED
|
|
execution.save()
|
|
latest = (
|
|
StepExecution.objects.filter(order=execution.order, step=target_step)
|
|
.order_by("-attempt")
|
|
.first()
|
|
)
|
|
attempt = (latest.attempt if latest else 0) + 1
|
|
return StepExecution.objects.create(
|
|
order=execution.order,
|
|
step=target_step,
|
|
state=StepExecution.State.READY,
|
|
attempt=attempt,
|
|
parent=execution,
|
|
)
|
|
|
|
@staticmethod
|
|
def _locked(execution) -> StepExecution:
|
|
return StepExecution.objects.select_for_update().get(pk=execution.pk)
|