Delete unused code in jobs

Author: vegito22

llmstack/jobs/jobs.py

@@ -10,9 +10,7 @@
 
 from llmstack.apps.apis import AppViewSet
 from llmstack.apps.models import App
-from llmstack.data.apis import DataSourceEntryViewSet
-from llmstack.data.models import DataSource
-from llmstack.jobs.models import AdhocJob, TaskRunLog, TaskStatus
+from llmstack.jobs.models import TaskRunLog, TaskStatus
 
 logger = logging.getLogger(__name__)
 
@@ -226,151 +224,3 @@ def run_app_task(app_id=None, input_data=None, *args, **kwargs):
     )
 
     return [SubTaskResult().model_dump()] * len(input_data)
-
-
-def upsert_datasource_entry_subtask(datasource_id, input_data):
-    request_input_data = {
-        "datasource_id": datasource_id,
-        "input_data": input_data,
-    }
-    datasource = DataSource.objects.get(uuid=uuid.UUID(datasource_id))
-    request = RequestFactory().post(
-        "/api/datasource_entries/upsert",
-        data=request_input_data,
-        format="json",
-    )
-    request.user = datasource.owner
-    request.data = request_input_data
-    response = DataSourceEntryViewSet().upsert(request)
-
-    return {
-        "status_code": response.status_code,
-        "data": response.data,
-    }
-
-
-def post_upsert_datasource_task(
-    task_run_log_uuid,
-    input_index,
-    status,
-    response,
-    job,
-):
-    if task_run_log_uuid:
-        task_run_log = TaskRunLog.objects.get(
-            uuid=uuid.UUID(task_run_log_uuid),
-        )
-        if status == TaskStatus.SUCCESS and response["status_code"] == 200:
-            task_run_log.result[input_index] = SubTaskResult(
-                status=TaskStatus.SUCCESS,
-                output="success",
-            ).model_dump()
-        else:
-            task_run_log.result[input_index] = SubTaskResult(
-                status=TaskStatus.FAILURE,
-                error=response,
-            ).model_dump()
-        task_run_log.save()
-
-    # If there are more input data to process, schedule the next task
-    input_data = job.meta["input_data"]
-    if input_index + 1 < len(input_data):
-        time_elapsed = (job.ended_at - job.started_at).total_seconds()
-        time_remaining_to_schedule_next_task = max(
-            (settings.TASK_RUN_DELAY - time_elapsed),
-            1,
-        )
-        logger.info(
-            f"Scheduling next task in {time_remaining_to_schedule_next_task} seconds",
-        )
-
-        django_rq.get_queue(job.meta["queue_name"]).enqueue(
-            upsert_datasource_entry_subtask,
-            args=(job.meta["datasource_id"], input_data[input_index + 1]),
-            on_success=upsert_datasource_entry_subtask_success_callback,
-            on_failure=upsert_datasource_entry_subtask_failure_callback,
-            meta={
-                "datasource_id": job.meta["datasource_id"],
-                "task_run_log_uuid": task_run_log_uuid,
-                "task_job_uuid": job.meta.get("task_job_uuid", None),
-                "input_data": input_data,
-                "input_data_index": input_index + 1,
-                "queue_name": job.meta["queue_name"],
-                "result_ttl": job.meta["result_ttl"],
-            },
-            result_ttl=job.meta["result_ttl"],
-        )
-    else:
-        # All tasks are completed. Update the task status to completed
-        task_job_uuid = job.meta.get("task_job_uuid", None)
-        if task_job_uuid:
-            job = AdhocJob.objects.get(uuid=uuid.UUID(task_job_uuid))
-            if job:
-                job.status = "finished"
-                job.save()
-
-
-def upsert_datasource_entry_subtask_success_callback(
-    job,
-    connection,
-    result,
-    *args,
-    **kwargs,
-):
-    try:
-        post_upsert_datasource_task(
-            job.meta["task_run_log_uuid"],
-            job.meta["input_data_index"],
-            TaskStatus.SUCCESS,
-            result,
-            job,
-        )
-    except Exception as e:
-        logger.error(f"Exception: {e}")
-
-
-def upsert_datasource_entry_subtask_failure_callback(
-    job,
-    connection,
-    type,
-    value,
-    traceback,
-):
-    logger.error(
-        f'task_run_log_uuid: {job.meta["task_run_log_uuid"]}, type: {type}, value: {value}, Traceback: {traceback} ',
-    )
-    try:
-        post_upsert_datasource_task(
-            job.meta["task_run_log_uuid"],
-            job.meta["input_data_index"],
-            TaskStatus.FAILURE,
-            f"Exception: {type}, detail: {value}",
-            job,
-        )
-    except Exception as e:
-        logger.error(f"Exception: {e}")
-
-
-def upsert_datasource_entries_task(datasource_id, input_data, *args, **kwargs):
-    job_metadata = kwargs["_job_metadata"]
-
-    result_ttl = 86400
-
-    django_rq.get_queue("default").enqueue(
-        upsert_datasource_entry_subtask,
-        args=(datasource_id, input_data[0]),
-        on_success=upsert_datasource_entry_subtask_success_callback,
-        on_failure=upsert_datasource_entry_subtask_failure_callback,
-        meta={
-            "datasource_id": datasource_id,
-            "task_run_log_uuid": job_metadata.get("task_run_log_uuid", None),
-            "task_job_uuid": job_metadata.get("task_job_uuid", None),
-            "input_data": input_data,
-            "input_data_index": 0,
-            "queue_name": "default",
-            "result_ttl": result_ttl,
-        },
-        result_ttl=result_ttl,
-    )
-
-    return [SubTaskResult().model_dump()] * len(input_data)