Merge pull request #251 from hydroserver2/299-celery-etl

299 celery etl

Author: kjlippold

api/urls.py

@@ -19,9 +19,10 @@
     datastream_router,
 )
 from etl.views import (
+    data_connection_router,
     orchestration_system_router,
-    data_source_router,
-    data_archive_router,
+    task_router,
+    task_run_router
 )
 
 
@@ -48,9 +49,10 @@
 api.add_router("processing-levels", processing_level_router)
 api.add_router("result-qualifiers", result_qualifier_router)
 
-api.add_router("orchestration-systems", orchestration_system_router)
-api.add_router("data-sources", data_source_router)
-api.add_router("data-archives", data_archive_router)
+api.add_router("etl-data-connections", data_connection_router)
+api.add_router("etl-tasks", task_router)
+api.add_router("etl-tasks", task_run_router)
+api.add_router("etl-orchestration-systems", orchestration_system_router)
 
 st_api_1_1 = SensorThingsAPI(
     title="HydroServer SensorThings API",

etl/admin.py

@@ -1,31 +1,58 @@
+from django.urls import path
 from django.contrib import admin
-from etl.models import OrchestrationSystem, DataSource, DataArchive
+from etl.models import OrchestrationSystem, DataConnection, Task, TaskMapping, TaskMappingPath, TaskRun
+from hydroserver.admin import VocabularyAdmin
 
 
-class OrchestrationSystemAdmin(admin.ModelAdmin):
+class OrchestrationSystemAdmin(admin.ModelAdmin, VocabularyAdmin):
     list_display = ("id", "name", "orchestration_system_type", "workspace__name")
+    change_list_template = "admin/etl/orchestrationsystem/change_list.html"
 
-    def delete_queryset(self, request, queryset):
-        OrchestrationSystem.delete_contents(filter_arg=queryset, filter_suffix="in")
-        queryset.delete()
+    def get_urls(self):
+        urls = super().get_urls()
 
+        return [
+            path(
+                "load-default-orchestration-system-data/",
+                self.admin_site.admin_view(self.load_default_data),
+                name="orchestration_system_load_default_data",
+            ),
+        ] + urls
 
-class DataSourceAdmin(admin.ModelAdmin):
-    list_display = ("id", "name", "orchestration_system__name", "workspace__name")
+    def load_default_data(self, request):
+        return self.load_fixtures(
+            request,
+            "admin:etl_orchestrationsystem_changelist",
+            ["etl/fixtures/default_orchestration_systems.yaml"],
+        )
 
-    def delete_queryset(self, request, queryset):
-        DataSource.delete_contents(filter_arg=queryset, filter_suffix="in")
-        queryset.delete()
 
+class DataConnectionAdmin(admin.ModelAdmin):
+    list_display = ("id", "name", "data_connection_type", "workspace__name")
 
-class DataArchiveAdmin(admin.ModelAdmin):
-    list_display = ("id", "name", "orchestration_system__name", "workspace__name")
 
-    def delete_queryset(self, request, queryset):
-        DataArchive.delete_contents(filter_arg=queryset, filter_suffix="in")
-        queryset.delete()
+class TaskAdmin(admin.ModelAdmin):
+    list_display = ("id", "name", "data_connection__name", "orchestration_system__name",
+                    "data_connection__workspace__name")
+
+
+class TaskMappingAdmin(admin.ModelAdmin):
+    list_display = ("id", "task__name", "source_identifier", "task__data_connection__name",
+                    "task__data_connection__workspace__name")
+
+
+class TaskMappingPathAdmin(admin.ModelAdmin):
+    list_display = ("id", "task_mapping__task__name", "target_identifier", "task_mapping__task__data_connection__name",
+                    "task_mapping__task__data_connection__workspace__name")
+
+
+class TaskRunAdmin(admin.ModelAdmin):
+    list_display = ("id", "status", "started_at", "finished_at", "result")
 
 
 admin.site.register(OrchestrationSystem, OrchestrationSystemAdmin)
-admin.site.register(DataSource, DataSourceAdmin)
-admin.site.register(DataArchive, DataArchiveAdmin)
+admin.site.register(DataConnection, DataConnectionAdmin)
+admin.site.register(Task, TaskAdmin)
+admin.site.register(TaskMapping, TaskMappingAdmin)
+admin.site.register(TaskMappingPath, TaskMappingPathAdmin)
+admin.site.register(TaskRun, TaskRunAdmin)

etl/fixtures/default_orchestration_systems.yaml

@@ -0,0 +1,7 @@
+---
+- model: etl.orchestrationsystem
+  pk: 019aeaa7-2310-7121-a627-e04427ff8bc9
+  fields:
+    name: HydroServer
+    orchestration_system_type: INTERNAL
+    workspace_id:

etl/fixtures/hydroshare_archival.yaml

@@ -0,0 +1,43 @@
+from datetime import timedelta
+from django.core.management.base import BaseCommand
+from django.utils import timezone
+from etl.models import Task, TaskRun
+
+
+class Command(BaseCommand):
+    help = "Removes old TaskRun records, keeping the most recent per Task."
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--days",
+            type=int,
+            default=14,
+            help="Number of days to keep TaskRun records. Older runs will be deleted. Default is 14.",
+        )
+
+    def handle(self, *args, **options):
+        days = options["days"]
+        now = timezone.now()
+        cutoff = now - timedelta(days=days)
+
+        total_deleted = 0
+
+        tasks = Task.objects.filter(
+            taskrun__started_at__lt=cutoff
+        ).distinct()
+
+        for task in tasks:
+            most_recent_run = TaskRun.objects.filter(task=task).order_by("-started_at").first()
+
+            deleted, _ = TaskRun.objects.filter(
+                task=task,
+                started_at__lt=cutoff
+            ).exclude(pk=most_recent_run.pk).delete()
+
+            total_deleted += deleted
+
+        self.stdout.write(
+            self.style.SUCCESS(
+                f"Cleanup complete. Deleted {total_deleted} old TaskRun records older than {days} days."
+            )
+        )

etl/management/commands/cleanup_etl_task_runs.py

@@ -7,21 +7,9 @@ class Command(BaseCommand):
 
     def handle(self, *args, **kwargs):
         fixtures = [
-            "tests/fixtures/test_users.yaml",
-            "tests/fixtures/test_workspaces.yaml",
-            "tests/fixtures/test_roles.yaml",
-            "tests/fixtures/test_collaborators.yaml",
-            "tests/fixtures/test_things.yaml",
-            "tests/fixtures/test_observed_properties.yaml",
-            "tests/fixtures/test_processing_levels.yaml",
-            "tests/fixtures/test_result_qualifiers.yaml",
-            "tests/fixtures/test_sensors.yaml",
-            "tests/fixtures/test_units.yaml",
-            "tests/fixtures/test_datastreams.yaml",
-            "tests/fixtures/test_orchestration_systems.yaml",
-            "tests/fixtures/test_data_sources.yaml",
-            "tests/fixtures/test_data_archives.yaml",
-            "tests/fixtures/test_etl_datastreams.yaml",
+            "tests/fixtures/test_etl_data_connections.yaml",
+            "tests/fixtures/test_etl_orchestration_systems.yaml",
+            "tests/fixtures/test_etl_tasks.yaml",
         ]
 
         for fixture in fixtures: