Rename Job model to JobProcess

Author: davegaeddert

plain-worker/plain/worker/admin.py

@@ -11,7 +11,7 @@
 from plain.http import ResponseRedirect
 from plain.runtime import settings
 
-from .models import Job, JobRequest, JobResult
+from .models import JobProcess, JobRequest, JobResult
 
 
 def _td_format(td_object):
@@ -87,14 +87,14 @@ class WaitingJobsCard(Card):
     title = "Waiting Jobs"
 
     def get_number(self):
-        return Job.query.waiting().count()
+        return JobProcess.query.waiting().count()
 
 
 class RunningJobsCard(Card):
     title = "Running Jobs"
 
     def get_number(self):
-        return Job.query.running().count()
+        return JobProcess.query.running().count()
 
 
 @register_viewset
@@ -117,11 +117,12 @@ class DetailView(AdminModelDetailView):
 
 
 @register_viewset
-class JobViewset(AdminViewset):
+class JobProcessViewset(AdminViewset):
     class ListView(AdminModelListView):
         nav_section = "Worker"
         nav_icon = "gear"
-        model = Job
+        model = JobProcess
+        title = "Job processes"
         fields = [
             "id",
             "job_class",
@@ -138,10 +139,10 @@ class ListView(AdminModelListView):
 
         def perform_action(self, action: str, target_ids: list):
             if action == "Delete":
-                Job.query.filter(id__in=target_ids).delete()
+                JobProcess.query.filter(id__in=target_ids).delete()
 
     class DetailView(AdminModelDetailView):
-        model = Job
+        model = JobProcess
 
 
 @register_viewset
@@ -162,7 +163,7 @@ class ListView(AdminModelListView):
         ]
         search_fields = [
             "uuid",
-            "job_uuid",
+            "job_process_uuid",
             "job_request_uuid",
             "job_class",
         ]

plain-worker/plain/worker/cli.py

@@ -8,7 +8,7 @@
 from plain.runtime import settings
 from plain.utils import timezone
 
-from .models import Job, JobRequest, JobResult
+from .models import JobProcess, JobRequest, JobResult
 from .registry import jobs_registry
 from .scheduling import load_schedule
 from .workers import Worker
@@ -100,7 +100,7 @@ def clear_completed():
 def stats():
     """Stats across all queues."""
     pending = JobRequest.query.count()
-    processing = Job.query.count()
+    processing = JobProcess.query.count()
 
     successful = JobResult.query.successful().count()
     errored = JobResult.query.errored().count()
@@ -124,7 +124,7 @@ def purge_processing():
     deleted = JobRequest.query.all().delete()[0]
     click.echo(f"Deleted {deleted} job requests")
 
-    deleted = Job.query.all().delete()[0]
+    deleted = JobProcess.query.all().delete()[0]
     click.echo(f"Deleted {deleted} jobs")
 
 

plain-worker/plain/worker/jobs.py

@@ -166,8 +166,8 @@ def run_in_worker(
                 return self._in_progress(unique_key)
 
     def _in_progress(self, unique_key):
-        """Get all JobRequests and Jobs that are currently in progress, regardless of queue."""
-        from .models import Job, JobRequest
+        """Get all JobRequests and JobProcess that are currently in progress, regardless of queue."""
+        from .models import JobProcess, JobRequest
 
         job_class_name = jobs_registry.get_job_class_name(self.__class__)
 
@@ -176,7 +176,7 @@ def _in_progress(self, unique_key):
             unique_key=unique_key,
         )
 
-        jobs = Job.query.filter(
+        jobs = JobProcess.query.filter(
             job_class=job_class_name,
             unique_key=unique_key,
         )

plain-worker/plain/worker/middleware.py

@@ -7,6 +7,6 @@ def __init__(self, run_job):
 
     def __call__(self, job):
         with app_logger.include_context(
-            job_request_uuid=str(job.job_request_uuid), job_uuid=str(job.uuid)
+            job_request_uuid=str(job.job_request_uuid), job_process_uuid=str(job.uuid)
         ):
             return self.run_job(job)

plain-worker/plain/worker/migrations/0003_rename_job_jobprocess_and_more.py

@@ -0,0 +1,67 @@
+# Generated by Plain 0.63.0 on 2025-09-15 21:29
+
+from plain import models
+from plain.models import migrations
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("plainworker", "0002_job_span_id_job_trace_id_jobrequest_span_id_and_more"),
+    ]
+
+    operations = [
+        migrations.RenameModel(
+            old_name="Job",
+            new_name="JobProcess",
+        ),
+        migrations.RemoveIndex(
+            model_name="jobresult",
+            name="plainworker_job_uui_8307d1_idx",
+        ),
+        migrations.RenameField(
+            model_name="jobresult",
+            old_name="job_uuid",
+            new_name="job_process_uuid",
+        ),
+        migrations.RenameIndex(
+            model_name="jobprocess",
+            new_name="plainworker_created_0d3928_idx",
+            old_name="plainworker_created_a02317_idx",
+        ),
+        migrations.RenameIndex(
+            model_name="jobprocess",
+            new_name="plainworker_queue_2550ba_idx",
+            old_name="plainworker_queue_077806_idx",
+        ),
+        migrations.RenameIndex(
+            model_name="jobprocess",
+            new_name="plainworker_unique__9dc0bb_idx",
+            old_name="plainworker_unique__04d87b_idx",
+        ),
+        migrations.RenameIndex(
+            model_name="jobprocess",
+            new_name="plainworker_started_b80ec5_idx",
+            old_name="plainworker_started_143df5_idx",
+        ),
+        migrations.RenameIndex(
+            model_name="jobprocess",
+            new_name="plainworker_job_cla_fe2b70_idx",
+            old_name="plainworker_job_cla_884b46_idx",
+        ),
+        migrations.RenameIndex(
+            model_name="jobprocess",
+            new_name="plainworker_job_req_357898_idx",
+            old_name="plainworker_job_req_db2681_idx",
+        ),
+        migrations.RenameIndex(
+            model_name="jobprocess",
+            new_name="plainworker_trace_i_da2cfa_idx",
+            old_name="plainworker_trace_i_d2b645_idx",
+        ),
+        migrations.AddIndex(
+            model_name="jobresult",
+            index=models.Index(
+                fields=["job_process_uuid"], name="plainworker_job_pro_ceabfb_idx"
+            ),
+        ),
+    ]

plain-worker/plain/worker/models.py

@@ -89,13 +89,13 @@ class Meta:
     def __str__(self):
         return f"{self.job_class} [{self.uuid}]"
 
-    def convert_to_job(self):
+    def convert_to_job_process(self):
         """
         JobRequests are the pending jobs that are waiting to be executed.
-        We immediately convert them to JobResults when they are picked up.
+        We immediately convert them to JobProcess when they are picked up.
         """
         with transaction.atomic():
-            result = Job.query.create(
+            result = JobProcess.query.create(
                 job_request_uuid=self.uuid,
                 job_class=self.job_class,
                 parameters=self.parameters,
@@ -140,7 +140,7 @@ def mark_lost_jobs(self):
 
 
 @models.register_model
-class Job(models.Model):
+class JobProcess(models.Model):
     """
     All active jobs are stored in this table.
     """
@@ -240,15 +240,15 @@ def run(self):
 
     def convert_to_result(self, *, status, error=""):
         """
-        Convert this Job to a JobResult.
+        Convert this JobProcess to a JobResult.
         """
         with transaction.atomic():
             result = JobResult.query.create(
                 ended_at=timezone.now(),
                 error=error,
                 status=status,
-                # From the Job
-                job_uuid=self.uuid,
+                # From the JobProcess
+                job_process_uuid=self.uuid,
                 started_at=self.started_at,
                 # From the JobRequest
                 job_request_uuid=self.job_request_uuid,
@@ -264,7 +264,7 @@ def convert_to_result(self, *, status, error=""):
                 span_id=self.span_id,
             )
 
-            # Delete the Job now
+            # Delete the JobProcess now
             self.delete()
 
         return result
@@ -359,7 +359,7 @@ class JobResult(models.Model):
     created_at = models.DateTimeField(auto_now_add=True)
 
     # From the Job
-    job_uuid = models.UUIDField()
+    job_process_uuid = models.UUIDField()
     started_at = models.DateTimeField(required=False, allow_null=True)
     ended_at = models.DateTimeField(required=False, allow_null=True)
     error = models.TextField(required=False)
@@ -391,7 +391,7 @@ class Meta:
         ordering = ["-created_at"]
         indexes = [
             models.Index(fields=["created_at"]),
-            models.Index(fields=["job_uuid"]),
+            models.Index(fields=["job_process_uuid"]),
             models.Index(fields=["started_at"]),
             models.Index(fields=["ended_at"]),
             models.Index(fields=["status"]),

plain-worker/plain/worker/workers.py

@@ -13,7 +13,7 @@
 from plain.utils import timezone
 from plain.utils.module_loading import import_string
 
-from .models import Job, JobRequest, JobResult, JobResultStatuses
+from .models import JobProcess, JobRequest, JobResult, JobResultStatuses
 from .registry import jobs_registry
 
 logger = logging.getLogger("plain.worker")
@@ -114,16 +114,18 @@ def run(self):
                     job_request.queue,
                 )
 
-                job = job_request.convert_to_job()
+                job = job_request.convert_to_job_process()
 
-            job_uuid = str(job.uuid)  # Make a str copy
+            job_process_uuid = str(job.uuid)  # Make a str copy
 
             # Release these now
             del job_request
             del job
 
-            future = self.executor.submit(process_job, job_uuid)
-            future.add_done_callback(partial(future_finished_callback, job_uuid))
+            future = self.executor.submit(process_job, job_process_uuid)
+            future.add_done_callback(
+                partial(future_finished_callback, job_process_uuid)
+            )
 
             # Do a quick sleep regardless to see if it
             # gives processes a chance to start up
@@ -213,7 +215,7 @@ def log_stats(self):
             num_proccesses = 0
 
         jobs_requested = JobRequest.query.filter(queue__in=self.queues).count()
-        jobs_processing = Job.query.filter(queue__in=self.queues).count()
+        jobs_processing = JobProcess.query.filter(queue__in=self.queues).count()
 
         logger.info(
             'Job worker stats worker_processes=%s worker_queues="%s" jobs_requested=%s jobs_processing=%s worker_max_processes=%s worker_max_jobs_per_process=%s',
@@ -228,52 +230,52 @@ def log_stats(self):
     def rescue_job_results(self):
         """Find any lost or failed jobs on this worker's queues and handle them."""
         # TODO return results and log them if there are any?
-        Job.query.filter(queue__in=self.queues).mark_lost_jobs()
+        JobProcess.query.filter(queue__in=self.queues).mark_lost_jobs()
         JobResult.query.filter(queue__in=self.queues).retry_failed_jobs()
 
 
-def future_finished_callback(job_uuid: str, future: Future):
+def future_finished_callback(job_process_uuid: str, future: Future):
     if future.cancelled():
-        logger.warning("Job cancelled job_uuid=%s", job_uuid)
+        logger.warning("Job cancelled job_process_uuid=%s", job_process_uuid)
         try:
-            job = Job.query.get(uuid=job_uuid)
+            job = JobProcess.query.get(uuid=job_process_uuid)
             job.convert_to_result(status=JobResultStatuses.CANCELLED)
-        except Job.DoesNotExist:
+        except JobProcess.DoesNotExist:
             # Job may have already been cleaned up
             pass
     elif exception := future.exception():
         # Process pool may have been killed...
         logger.warning(
-            "Job failed job_uuid=%s",
-            job_uuid,
+            "Job failed job_process_uuid=%s",
+            job_process_uuid,
             exc_info=exception,
         )
         try:
-            job = Job.query.get(uuid=job_uuid)
+            job = JobProcess.query.get(uuid=job_process_uuid)
             job.convert_to_result(status=JobResultStatuses.CANCELLED)
-        except Job.DoesNotExist:
+        except JobProcess.DoesNotExist:
             # Job may have already been cleaned up
             pass
     else:
-        logger.debug("Job finished job_uuid=%s", job_uuid)
+        logger.debug("Job finished job_process_uuid=%s", job_process_uuid)
 
 
-def process_job(job_uuid):
+def process_job(job_process_uuid):
     try:
         worker_pid = os.getpid()
 
         request_started.send(sender=None)
 
-        job = Job.query.get(uuid=job_uuid)
+        job_process = JobProcess.query.get(uuid=job_process_uuid)
 
         logger.info(
             'Executing job worker_pid=%s job_class=%s job_request_uuid=%s job_priority=%s job_source="%s" job_queue="%s"',
             worker_pid,
-            job.job_class,
-            job.job_request_uuid,
-            job.priority,
-            job.source,
-            job.queue,
+            job_process.job_class,
+            job_process.job_request_uuid,
+            job_process.priority,
+            job_process.source,
+            job_process.queue,
         )
 
         def middleware_chain(job):
@@ -284,19 +286,19 @@ def middleware_chain(job):
             middleware_instance = middleware_class(middleware_chain)
             middleware_chain = middleware_instance
 
-        job_result = middleware_chain(job)
+        job_result = middleware_chain(job_process)
 
         # Release it now
-        del job
+        del job_process
 
         duration = job_result.ended_at - job_result.started_at
         duration = duration.total_seconds()
 
         logger.info(
-            'Completed job worker_pid=%s job_class=%s job_uuid=%s job_request_uuid=%s job_result_uuid=%s job_priority=%s job_source="%s" job_queue="%s" job_duration=%s',
+            'Completed job worker_pid=%s job_class=%s job_process_uuid=%s job_request_uuid=%s job_result_uuid=%s job_priority=%s job_source="%s" job_queue="%s" job_duration=%s',
             worker_pid,
             job_result.job_class,
-            job_result.job_uuid,
+            job_result.job_process_uuid,
             job_result.job_request_uuid,
             job_result.uuid,
             job_result.priority,