Fixed old references to payloads

Author: Ken Lippold

src/hydroserverpy/etl/etl_configuration.py

@@ -85,9 +85,9 @@ def check_timezone(cls, timezone_value, info):
         return timezone_value
 
 
-class PerPayloadPlaceholder(BaseModel):
+class PerTaskPlaceholder(BaseModel):
     name: str
-    type: Literal["perPayload"]
+    type: Literal["perTask"]
 
 
 class RunTimePlaceholder(BaseModel):
@@ -101,7 +101,7 @@ class Config:
 
 
 PlaceholderVariable = Annotated[
-    Union[PerPayloadPlaceholder, RunTimePlaceholder],
+    Union[PerTaskPlaceholder, RunTimePlaceholder],
     Field(discriminator="type"),
 ]
 
@@ -206,7 +206,7 @@ class Config:
         populate_by_name = True
 
 
-class Payload(BaseModel):
+class Task(BaseModel):
     uid: uuid.UUID = Field(..., alias="id")
     name: str = ""
     mappings: List[SourceTargetMapping] = Field(default_factory=list)
@@ -222,13 +222,3 @@ class Payload(BaseModel):
 
     class Config:
         populate_by_name = True
-
-
-class EtlConfiguration(BaseModel):
-    uid: uuid.UUID = Field(..., alias="id")
-    name: str
-    type: WorkflowType
-    extractor: ExtractorConfig
-    transformer: TransformerConfig
-    loader: LoaderConfig
-    payloads: List[Payload]

src/hydroserverpy/etl/extractors/base.py

@@ -2,15 +2,15 @@
 import logging
 import pandas as pd
 from datetime import datetime
-from ..etl_configuration import ExtractorConfig, Payload
+from ..etl_configuration import ExtractorConfig, Task
 from ..timestamp_parser import TimestampParser
 
 
 class Extractor:
     def __init__(self, extractor_config: ExtractorConfig):
         self.cfg = extractor_config
 
-    def resolve_placeholder_variables(self, payload: Payload, loader):
+    def resolve_placeholder_variables(self, task: Task, loader):
         logging.info(f"Creating runtime variables...")
         filled = {}
         for placeholder in self.cfg.placeholder_variables:
@@ -19,14 +19,14 @@ def resolve_placeholder_variables(self, payload: Payload, loader):
             if placeholder.type == "runTime":
                 logging.info(f"Resolving runtime var: {name}")
                 if placeholder.run_time_value == "latestObservationTimestamp":
-                    value = loader.earliest_begin_date(payload)
+                    value = loader.earliest_begin_date(task)
                 elif placeholder.run_time_value == "jobExecutionTime":
                     value = pd.Timestamp.now(tz="UTC")
-            elif placeholder.type == "perPayload":
-                logging.info(f"Resolving payload var: {name}")
-                if name not in payload.extractor_variables:
-                    raise KeyError(f"Missing per-payload variable '{name}'")
-                value = payload.extractor_variables[name]
+            elif placeholder.type == "perTask":
+                logging.info(f"Resolving task var: {name}")
+                if name not in task.extractor_variables:
+                    raise KeyError(f"Missing per-task variable '{name}'")
+                value = task.extractor_variables[name]
             else:
                 continue
 

src/hydroserverpy/etl/extractors/http_extractor.py

@@ -2,19 +2,19 @@
 import requests
 from io import BytesIO
 
-from ..etl_configuration import Payload
+from ..etl_configuration import Task
 from .base import Extractor, ExtractorConfig
 
 
 class HTTPExtractor(Extractor):
     def __init__(self, settings: ExtractorConfig):
         super().__init__(settings)
 
-    def extract(self, payload: Payload, loader=None):
+    def extract(self, task: Task, loader=None):
         """
         Downloads the file from the HTTP/HTTPS server and returns a file-like object.
         """
-        url = self.resolve_placeholder_variables(payload, loader)
+        url = self.resolve_placeholder_variables(task, loader)
         logging.info(f"Requesting data from → {url}")
 
         response = requests.get(url)

src/hydroserverpy/etl/loaders/base.py

@@ -7,5 +7,5 @@ def load(self, *args, **kwargs) -> None:
         pass
 
     @abstractmethod
-    def earliest_begin_date(self, payload_mappings) -> str:
+    def earliest_begin_date(self, task_mappings) -> str:
         pass

src/hydroserverpy/etl/loaders/hydroserver_loader.py

@@ -4,7 +4,7 @@
 from .base import Loader
 import logging
 import pandas as pd
-from ..etl_configuration import Payload, SourceTargetMapping
+from ..etl_configuration import Task, SourceTargetMapping
 
 if TYPE_CHECKING:
     from hydroserverpy.api.client import HydroServer
@@ -20,12 +20,12 @@ def __init__(self, client: HydroServer, task_id):
         self._begin_cache: dict[str, pd.Timestamp] = {}
         self.task_id = task_id
 
-    def load(self, data: pd.DataFrame, payload: Payload) -> None:
+    def load(self, data: pd.DataFrame, task: Task) -> None:
         """
         Load observations from a DataFrame to the HydroServer.
         :param data: A Pandas DataFrame where each column corresponds to a datastream.
         """
-        begin_date = self.earliest_begin_date(payload)
+        begin_date = self.earliest_begin_date(task)
         new_data = data[data["timestamp"] > begin_date]
         for col in new_data.columns.difference(["timestamp"]):
             df = (
@@ -73,7 +73,7 @@ def load(self, data: pd.DataFrame, payload: Payload) -> None:
     def _fetch_earliest_begin(
         self, mappings: list[SourceTargetMapping]
     ) -> pd.Timestamp:
-        logging.info("Querying HydroServer for earliest begin date for payload...")
+        logging.info("Querying HydroServer for earliest begin date for task...")
         timestamps = []
         datastreams = self.client.datastreams.list(
             data_source=self.data_source_id
@@ -88,11 +88,11 @@ def _fetch_earliest_begin(
         logging.info(f"Found earliest begin date: {min(timestamps)}")
         return min(timestamps)
 
-    def earliest_begin_date(self, payload: Payload) -> pd.Timestamp:
+    def earliest_begin_date(self, task: Task) -> pd.Timestamp:
         """
-        Return earliest begin date for a payload, or compute+cache it on first call.
+        Return earliest begin date for a task, or compute+cache it on first call.
         """
-        key = payload.name
+        key = task.name
         if key not in self._begin_cache:
-            self._begin_cache[key] = self._fetch_earliest_begin(payload.mappings)
+            self._begin_cache[key] = self._fetch_earliest_begin(task.mappings)
         return self._begin_cache[key]

src/hydroserverpy/etl/transformers/base.py

@@ -69,9 +69,9 @@ def standardize_dataframe(
         self, df: pd.DataFrame, mappings: List[SourceTargetMapping]
     ):
         if not df.empty:
-            logging.info(f"Read payload into dataframe: {df.iloc[0].to_dict()}")
+            logging.info(f"Read task into dataframe: {df.iloc[0].to_dict()}")
         else:
-            logging.info("Read payload into dataframe: [empty dataframe]")
+            logging.info("Read task into dataframe: [empty dataframe]")
 
         # 1) Normalize timestamp column
         df.rename(columns={self.timestamp.key: "timestamp"}, inplace=True)