Merge pull request #23 from AbsaOSS/Release/2.0.3

added timestamp to bookkeeper

CHANGELOG.md

@@ -1,6 +1,10 @@
 # Change Log
 All notable changes to this project will be documented in this file.
 
+## 2.0.3 - 2025-01-27
+  ### Runner
+  - added run time timestamp to record
+
 ## 2.0.2 - 2025-01-24
   ### Runner
   - new bookkeeping functionality

docs/source/conf.py

@@ -28,7 +28,7 @@
 project = "rialto"
 copyright = "2022, Marek Dobransky"
 author = "Marek Dobransky"
-release = "2.0.2"
+release = "2.0.3"
 
 # -- General configuration ---------------------------------------------------
 # https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration

pyproject.toml

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "rialto"
 
-version = "2.0.2"
+version = "2.0.3"
 
 packages = [
     { include = "rialto" },

rialto/runner/bookkeeper.py

@@ -38,7 +38,7 @@ def add(self, record: Record) -> None:
 
         :param record: Record to add to the table.
         """
-        new = self.spark.createDataFrame([record.to_spark_row()], Record.schema)
+        new = self.spark.createDataFrame([record.to_spark_row()], Record.get_schema())
         db = self._load()
         if db:
             db = db.unionByName(new)

rialto/runner/record.py

@@ -16,10 +16,17 @@
 
 from dataclasses import dataclass
 from datetime import datetime, timedelta
-from typing import ClassVar, Optional
+from typing import Optional
 
 from pyspark.sql import Row
-from pyspark.sql.types import DateType, IntegerType, StringType, StructField, StructType
+from pyspark.sql.types import (
+    DateType,
+    IntegerType,
+    StringType,
+    StructField,
+    StructType,
+    TimestampType,
+)
 
 
 @dataclass
@@ -34,19 +41,23 @@ class Record:
     status: str
     reason: str
     exception: Optional[str] = None
+    run_timestamp: datetime.timestamp = datetime.now().isoformat(sep=" ", timespec="seconds")
 
-    schema: ClassVar[StructType] = StructType(
-        [
-            StructField("job", StringType(), nullable=False),
-            StructField("target", StringType(), nullable=False),
-            StructField("date", DateType(), nullable=False),
-            StructField("time", StringType(), nullable=False),
-            StructField("records", IntegerType(), nullable=False),
-            StructField("status", StringType(), nullable=False),
-            StructField("reason", StringType(), nullable=False),
-            StructField("exception", StringType(), nullable=True),
-        ]
-    )
+    def get_schema(self) -> StructType:
+        """Retrieve schema of pyspark DataFrame"""
+        return StructType(
+            [
+                StructField("job", StringType(), nullable=False),
+                StructField("target", StringType(), nullable=False),
+                StructField("date", DateType(), nullable=False),
+                StructField("time", StringType(), nullable=False),
+                StructField("records", IntegerType(), nullable=False),
+                StructField("status", StringType(), nullable=False),
+                StructField("reason", StringType(), nullable=False),
+                StructField("exception", StringType(), nullable=True),
+                StructField("run_timestamp", TimestampType(), nullable=False),
+            ]
+        )
 
     def to_spark_row(self) -> Row:
         """Convert Record to Spark Row"""
@@ -59,4 +70,5 @@ def to_spark_row(self) -> Row:
             status=self.status,
             reason=self.reason,
             exception=self.exception,
+            run_timestamp=self.run_timestamp,
         )

tests/runner/test_bookkeeping.py

@@ -1,4 +1,4 @@
-from datetime import timedelta
+from datetime import datetime, timedelta
 
 from rialto.runner.date_manager import DateManager
 from rialto.runner.record import Record
@@ -11,6 +11,8 @@
     1,
     "status",
     "reason",
+    None,
+    datetime(2024, 1, 1, 1, 2, 3),
 )
 
 
@@ -24,3 +26,4 @@ def test_record_to_spark(spark):
     assert row.status == "status"
     assert row.reason == "reason"
     assert row.exception is None
+    assert row.run_timestamp == datetime(2024, 1, 1, 1, 2, 3)