Added updated ETL models and endpoints

Author: Ken Lippold

etl/admin.py

@@ -1,3 +1,9 @@
 from django.contrib import admin
+from etl.models import EtlSystemPlatform, EtlSystem, EtlConfiguration, DataSource, LinkedDatastream
 
-# Register your models here.
+
+admin.site.register(EtlSystemPlatform)
+admin.site.register(EtlSystem)
+admin.site.register(EtlConfiguration)
+admin.site.register(DataSource)
+admin.site.register(LinkedDatastream)

etl/management/__init__.py

@@ -0,0 +1,33 @@
+from django.core.management.base import BaseCommand
+from django.core.management import call_command
+
+
+class Command(BaseCommand):
+    help = "Loads test data from fixtures"
+
+    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_etl_system_platforms.yaml",
+            "tests/fixtures/test_etl_configurations.yaml",
+            "tests/fixtures/test_etl_systems.yaml",
+            "tests/fixtures/test_data_sources.yaml",
+        ]
+
+        for fixture in fixtures:
+            self.stdout.write(self.style.NOTICE(f"Loading fixture: {fixture}"))
+            try:
+                call_command("loaddata", fixture)
+                self.stdout.write(self.style.SUCCESS(f"Successfully loaded {fixture}"))
+            except Exception as e:
+                self.stderr.write(self.style.ERROR(f"Failed to load {fixture}: {e}"))

etl/management/commands/__init__.py

@@ -0,0 +1,91 @@
+# Generated by Django 5.2b1 on 2025-03-11 22:59
+
+import django.db.models.deletion
+import iam.models.utils
+import uuid
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ('iam', '0002_load_default_data'),
+        ('sta', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='EtlConfiguration',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
+                ('name', models.CharField(max_length=255)),
+                ('etl_configuration_type', models.CharField(max_length=255)),
+                ('etl_configuration_schema', models.JSONField()),
+            ],
+            bases=(models.Model, iam.models.utils.PermissionChecker),
+        ),
+        migrations.CreateModel(
+            name='EtlSystem',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
+                ('name', models.CharField(max_length=255)),
+                ('workspace', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.DO_NOTHING, related_name='etl_systems', to='iam.workspace')),
+            ],
+            bases=(models.Model, iam.models.utils.PermissionChecker),
+        ),
+        migrations.CreateModel(
+            name='DataSource',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
+                ('name', models.CharField(max_length=255)),
+                ('interval', models.PositiveIntegerField(blank=True, null=True)),
+                ('interval_units', models.CharField(blank=True, max_length=255, null=True)),
+                ('crontab', models.CharField(blank=True, max_length=255, null=True)),
+                ('start_time', models.DateTimeField(blank=True, null=True)),
+                ('end_time', models.DateTimeField(blank=True, null=True)),
+                ('paused', models.BooleanField(default=False)),
+                ('last_run_successful', models.BooleanField(blank=True, null=True)),
+                ('last_run_message', models.TextField(blank=True, null=True)),
+                ('last_run', models.DateTimeField(blank=True, null=True)),
+                ('next_run', models.DateTimeField(blank=True, null=True)),
+                ('etl_configuration_settings', models.JSONField(blank=True, null=True)),
+                ('workspace', models.ForeignKey(on_delete=django.db.models.deletion.DO_NOTHING, related_name='data_sources', to='iam.workspace')),
+                ('etl_configuration', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.DO_NOTHING, related_name='data_sources', to='etl.etlconfiguration')),
+                ('etl_system', models.ForeignKey(on_delete=django.db.models.deletion.DO_NOTHING, related_name='data_sources', to='etl.etlsystem')),
+            ],
+            bases=(models.Model, iam.models.utils.PermissionChecker),
+        ),
+        migrations.CreateModel(
+            name='EtlSystemPlatform',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
+                ('name', models.CharField(max_length=255)),
+                ('interval_schedule_supported', models.BooleanField(default=False)),
+                ('crontab_schedule_supported', models.BooleanField(default=False)),
+                ('workspace', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.DO_NOTHING, related_name='etl_system_platforms', to='iam.workspace')),
+            ],
+            bases=(models.Model, iam.models.utils.PermissionChecker),
+        ),
+        migrations.AddField(
+            model_name='etlsystem',
+            name='etl_system_platform',
+            field=models.ForeignKey(on_delete=django.db.models.deletion.DO_NOTHING, related_name='etl_systems', to='etl.etlsystemplatform'),
+        ),
+        migrations.AddField(
+            model_name='etlconfiguration',
+            name='etl_system_platform',
+            field=models.ForeignKey(on_delete=django.db.models.deletion.DO_NOTHING, related_name='etl_configurations', to='etl.etlsystemplatform'),
+        ),
+        migrations.CreateModel(
+            name='LinkedDatastream',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('etl_configuration_settings', models.JSONField(blank=True, null=True)),
+                ('data_source', models.ForeignKey(on_delete=django.db.models.deletion.DO_NOTHING, related_name='linked_datastreams', to='etl.datasource')),
+                ('datastream', models.OneToOneField(on_delete=django.db.models.deletion.DO_NOTHING, related_name='data_source', to='sta.datastream')),
+                ('etl_configuration', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.DO_NOTHING, related_name='linked_datastreams', to='etl.etlconfiguration')),
+            ],
+        ),
+    ]

etl/management/commands/load_etl_test_data.py

@@ -0,0 +1,4 @@
+from .data_source import DataSource, LinkedDatastream
+from .etl_system_platform import EtlSystemPlatform
+from .etl_system import EtlSystem
+from .etl_configuration import EtlConfiguration

etl/migrations/0001_initial.py

@@ -0,0 +1,78 @@
+import uuid
+import typing
+from typing import Literal, Optional
+from django.db import models
+from django.db.models import Q
+from iam.models import Workspace
+from iam.models.utils import PermissionChecker
+from sta.models import Datastream
+from .etl_system import EtlSystem
+from .etl_configuration import EtlConfiguration
+
+if typing.TYPE_CHECKING:
+    from django.contrib.auth import get_user_model
+    from iam.models import Workspace
+
+    User = get_user_model()
+
+
+class DataSourceQuerySet(models.QuerySet):
+    def visible(self, user: "User"):
+        if user.account_type == "admin":
+            return self
+        else:
+            return self.filter(
+                Q(workspace__owner=user) |
+                Q(workspace__collaborators__user=user,
+                  workspace__collaborators__role__permissions__resource_type__in=["*", "DataSource"],
+                  workspace__collaborators__role__permissions__permission_type__in=["*", "view"])
+            )
+
+
+class DataSource(models.Model, PermissionChecker):
+    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+    workspace = models.ForeignKey(Workspace, related_name="data_sources", on_delete=models.DO_NOTHING)
+    name = models.CharField(max_length=255)
+    etl_system = models.ForeignKey(EtlSystem, related_name="data_sources", on_delete=models.DO_NOTHING)
+    interval = models.PositiveIntegerField(blank=True, null=True)
+    interval_units = models.CharField(max_length=255, blank=True, null=True)
+    crontab = models.CharField(max_length=255, blank=True, null=True)
+    start_time = models.DateTimeField(blank=True, null=True)
+    end_time = models.DateTimeField(blank=True, null=True)
+    paused = models.BooleanField(default=False)
+    last_run_successful = models.BooleanField(blank=True, null=True)
+    last_run_message = models.TextField(blank=True, null=True)
+    last_run = models.DateTimeField(blank=True, null=True)
+    next_run = models.DateTimeField(blank=True, null=True)
+    etl_configuration = models.ForeignKey(EtlConfiguration, related_name="data_sources", on_delete=models.DO_NOTHING,
+                                          blank=True, null=True)
+    etl_configuration_settings = models.JSONField(blank=True, null=True)
+
+    objects = DataSourceQuerySet.as_manager()
+
+    @classmethod
+    def can_user_create(cls, user: Optional["User"], workspace: "Workspace"):
+        return cls.check_create_permissions(user=user, workspace=workspace, resource_type="DataSource")
+
+    def get_user_permissions(self, user: Optional["User"]) -> list[Literal["edit", "delete", "view"]]:
+        user_permissions = self.check_object_permissions(user=user, workspace=self.workspace,
+                                                         resource_type="DataSource")
+
+        return user_permissions
+
+    def delete(self, *args, **kwargs):
+        self.delete_contents(filter_arg=self, filter_suffix="")
+        super().delete(*args, **kwargs)
+
+    @staticmethod
+    def delete_contents(filter_arg: models.Model, filter_suffix: Optional[str]):
+        data_source_relation_filter = f"data_source__{filter_suffix}" if filter_suffix else "data_source"
+        LinkedDatastream.objects.filter(**{data_source_relation_filter: filter_arg}).delete()
+
+
+class LinkedDatastream(models.Model):
+    data_source = models.ForeignKey(DataSource, related_name="linked_datastreams", on_delete=models.DO_NOTHING)
+    datastream = models.OneToOneField(Datastream, related_name="data_source", on_delete=models.DO_NOTHING)
+    etl_configuration = models.ForeignKey(EtlConfiguration, related_name="linked_datastreams",
+                                          on_delete=models.DO_NOTHING, blank=True, null=True)
+    etl_configuration_settings = models.JSONField(blank=True, null=True)

etl/models/__init__.py

@@ -0,0 +1,73 @@
+import uuid
+import typing
+from typing import Literal, Optional
+from django.db import models
+from django.db.models import Q
+from iam.models import Workspace
+from iam.models.utils import PermissionChecker
+from .etl_system_platform import EtlSystemPlatform
+
+if typing.TYPE_CHECKING:
+    from django.contrib.auth import get_user_model
+    from iam.models import Workspace
+
+    User = get_user_model()
+
+
+class EtlConfigurationQuerySet(models.QuerySet):
+    def visible(self, user: Optional["User"]):
+        if user is None:
+            return self.filter(
+                Q(etl_system_platform__workspace__isnull=True)
+            )
+        elif user.account_type == "admin":
+            return self
+        else:
+            return self.filter(
+                Q(etl_system_platform__workspace__isnull=True) |
+                Q(etl_system_platform__workspace__owner=user) |
+                Q(etl_system_platform__workspace__collaborators__user=user,
+                  etl_system_platform__workspace__collaborators__role__permissions__resource_type__in=[
+                      "*", "EtlConfiguration"
+                  ],
+                  etl_system_platform__workspace__collaborators__role__permissions__permission_type__in=["*", "view"])
+            )
+
+
+class EtlConfiguration(models.Model, PermissionChecker):
+    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+    etl_system_platform = models.ForeignKey(EtlSystemPlatform, related_name="etl_configurations",
+                                            on_delete=models.DO_NOTHING)
+    name = models.CharField(max_length=255)
+    etl_configuration_type = models.CharField(max_length=255)
+    etl_configuration_schema = models.JSONField()
+
+    objects = EtlConfigurationQuerySet.as_manager()
+
+    @classmethod
+    def can_user_create(cls, user: Optional["User"], workspace: "Workspace"):
+        return cls.check_create_permissions(user=user, workspace=workspace, resource_type="EtlConfiguration")
+
+    def get_user_permissions(self, user: Optional["User"]) -> list[Literal["edit", "delete", "view"]]:
+        user_permissions = self.check_object_permissions(user=user, workspace=self.etl_system_platform.workspace,
+                                                         resource_type="EtlConfiguration")
+
+        return user_permissions
+
+    def delete(self, *args, **kwargs):
+        self.delete_contents(filter_arg=self, filter_suffix="")
+        super().delete(*args, **kwargs)
+
+    @staticmethod
+    def delete_contents(filter_arg: models.Model, filter_suffix: Optional[str]):
+        from etl.models import DataSource, LinkedDatastream
+
+        etl_configuration_relation_filter = f"etl_configuration__{filter_suffix}" \
+            if filter_suffix else "etl_configuration"
+
+        DataSource.objects.filter(
+            **{etl_configuration_relation_filter: filter_arg}
+        ).update(etl_configuration=None)
+        LinkedDatastream.objects.filter(
+            **{etl_configuration_relation_filter: filter_arg}
+        ).update(etl_configuration=None)

etl/models/data_source.py

@@ -0,0 +1,65 @@
+import uuid
+import typing
+from typing import Literal, Optional
+from django.db import models
+from django.db.models import Q
+from iam.models import Workspace
+from iam.models.utils import PermissionChecker
+from .etl_system_platform import EtlSystemPlatform
+
+if typing.TYPE_CHECKING:
+    from django.contrib.auth import get_user_model
+    from iam.models import Workspace
+
+    User = get_user_model()
+
+
+class EtlSystemQuerySet(models.QuerySet):
+    def visible(self, user: Optional["User"]):
+        if user is None:
+            return self.filter(
+                Q(workspace__isnull=True)
+            )
+        elif user.account_type == "admin":
+            return self
+        else:
+            return self.filter(
+                Q(workspace__isnull=True) |
+                Q(workspace__owner=user) |
+                Q(workspace__collaborators__user=user,
+                  workspace__collaborators__role__permissions__resource_type__in=["*", "EtlSystem"],
+                  workspace__collaborators__role__permissions__permission_type__in=["*", "view"])
+            )
+
+
+class EtlSystem(models.Model, PermissionChecker):
+    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+    workspace = models.ForeignKey(Workspace, related_name="etl_systems", on_delete=models.DO_NOTHING, blank=True,
+                                  null=True)
+    name = models.CharField(max_length=255)
+    etl_system_platform = models.ForeignKey(EtlSystemPlatform, related_name="etl_systems", on_delete=models.DO_NOTHING)
+
+    objects = EtlSystemQuerySet.as_manager()
+
+    @classmethod
+    def can_user_create(cls, user: Optional["User"], workspace: "Workspace"):
+        return cls.check_create_permissions(user=user, workspace=workspace, resource_type="EtlSystem")
+
+    def get_user_permissions(self, user: Optional["User"]) -> list[Literal["edit", "delete", "view"]]:
+        user_permissions = self.check_object_permissions(user=user, workspace=self.workspace,
+                                                         resource_type="EtlSystem")
+
+        return user_permissions
+
+    def delete(self, *args, **kwargs):
+        self.delete_contents(filter_arg=self, filter_suffix="")
+        super().delete(*args, **kwargs)
+
+    @staticmethod
+    def delete_contents(filter_arg: models.Model, filter_suffix: Optional[str]):
+        from etl.models import DataSource
+
+        etl_system_relation_filter = f"etl_system__{filter_suffix}" if filter_suffix else "etl_system"
+
+        DataSource.delete_contents(filter_arg=filter_arg, filter_suffix=etl_system_relation_filter)
+        DataSource.objects.filter(**{etl_system_relation_filter: filter_arg}).delete()