299 celery etl (#20)

* Refactored etl module to use jobs and tasks API

* Fixed old references to payloads

* Uptick version

* Renamed Job to DataConnection

* Added ETL local exec option

* Updated version

---------

Co-authored-by: Ken Lippold <klippold@Kens-MacBook-Pro-2.local>

Author: kjlippold

setup.cfg

@@ -19,7 +19,7 @@ install_requires =
     numpy >= 1.22.4
     pyyaml >= 5
     simplejson >= 3
-    crontab >= 1
+    python-crontab >= 3
     python-dateutil >= 2.8.2
     croniter >= 2.0.1
     jmespath >= 1.0.1

src/hydroserverpy/api/client.py

@@ -12,8 +12,8 @@
     SensorService,
     DatastreamService,
     OrchestrationSystemService,
-    DataSourceService,
-    DataArchiveService,
+    DataConnectionService,
+    TaskService,
 )
 
 
@@ -191,13 +191,13 @@ def orchestrationsystems(self):
         return OrchestrationSystemService(self)
 
     @property
-    def datasources(self):
-        """Utilities for managing HydroServer data sources."""
+    def dataconnections(self):
+        """Utilities for managing HydroServer ETL data connections."""
 
-        return DataSourceService(self)
+        return DataConnectionService(self)
 
     @property
-    def dataarchives(self):
-        """Utilities for managing HydroServer data archives."""
+    def tasks(self):
+        """Utilities for managing HydroServer ETL tasks."""
 
-        return DataArchiveService(self)
+        return TaskService(self)

src/hydroserverpy/api/models/__init__.py

@@ -13,8 +13,10 @@
 from .sta.thing import Thing
 from .sta.unit import Unit
 from .etl.orchestration_system import OrchestrationSystem
-from .etl.data_source import DataSource
-from .etl.data_archive import DataArchive
+from .etl.data_connection import DataConnection
+from .etl.run import TaskRun
+from .etl.task import Task
+from .etl.mapping import TaskMapping
 
 Workspace.model_rebuild()
 Role.model_rebuild()

src/hydroserverpy/api/models/etl/__init__.py

@@ -1,26 +1,13 @@
-from .extractors import Extractor, HTTPExtractor, LocalFileExtractor, FTPExtractor
-from .transformers import JSONTransformer, CSVTransformer, Transformer
-from .loaders import HydroServerLoader, Loader
-
-from .etl_configuration import EtlConfiguration
-from .schedule import Schedule
-from .status import Status
 from .orchestration_system import OrchestrationSystem
-from .data_source import DataSource
+from .data_connection import DataConnection
+from .task import Task
+from .schedule import TaskSchedule
+from .run import TaskRun
 
 __all__ = [
-    "CSVTransformer",
-    "JSONTransformer",
-    "LocalFileExtractor",
-    "FTPExtractor",
-    "HTTPExtractor",
-    "Extractor",
-    "Transformer",
-    "Loader",
-    "HydroServerLoader",
-    "EtlConfiguration",
-    "Schedule",
-    "Status",
     "OrchestrationSystem",
-    "DataSource",
+    "DataConnection",
+    "Task",
+    "TaskSchedule",
+    "TaskRun"
 ]

src/hydroserverpy/api/models/etl/data_archive.py

@@ -0,0 +1,63 @@
+import uuid
+from typing import ClassVar, List, Optional, TYPE_CHECKING
+from pydantic import Field, AliasPath, AliasChoices
+from ..base import HydroServerBaseModel
+
+if TYPE_CHECKING:
+    from hydroserverpy import HydroServer
+    from hydroserverpy.api.models import Workspace, Task
+
+
+class DataConnection(HydroServerBaseModel):
+    name: str = Field(..., max_length=255)
+    data_connection_type: str = Field(..., max_length=255, alias="type")
+    workspace_id: Optional[uuid.UUID] = Field(
+        None, validation_alias=AliasChoices("workspaceId", AliasPath("workspace", "id"))
+    )
+    extractor_type: str = Field(..., max_length=255, validation_alias=AliasPath("extractor", "type"))
+    extractor_settings: dict = Field(default_factory=dict, validation_alias=AliasPath("extractor", "settings"))
+    transformer_type: str = Field(..., max_length=255, validation_alias=AliasPath("transformer", "type"))
+    transformer_settings: dict = Field(default_factory=dict, validation_alias=AliasPath("transformer", "settings"))
+    loader_type: str = Field(..., max_length=255, validation_alias=AliasPath("loader", "type"))
+    loader_settings: dict = Field(default_factory=dict, validation_alias=AliasPath("loader", "settings"))
+
+    _editable_fields: ClassVar[set[str]] = {
+        "name",
+        "data_connection_type",
+        "extractor_type",
+        "extractor_settings",
+        "transformer_type",
+        "transformer_settings",
+        "loader_type",
+        "loader_settings",
+    }
+
+    def __init__(self, client: "HydroServer", **data):
+        super().__init__(client=client, service=client.dataconnections, **data)
+
+        self._workspace = None
+        self._tasks = None
+
+    @classmethod
+    def get_route(cls):
+        return "etl-data-connections"
+
+    @property
+    def workspace(self) -> "Workspace":
+        """The workspace this ETL data connection belongs to."""
+
+        if self._workspace is None and self.workspace_id:
+            self._workspace = self.client.workspaces.get(uid=self.workspace_id)
+
+        return self._workspace
+
+    @property
+    def tasks(self) -> List["Task"]:
+        """The ETL tasks associated with this ETL data connection."""
+
+        if self._tasks is None:
+            self._tasks = self.client.tasks.list(
+                data_connection=self.uid, fetch_all=True
+            ).items
+
+        return self._tasks