add dataversion to sensor result

TEAMSchools · Aug 16, 2023 · 8f4071e · 8f4071e
1 parent b50269d
commit 8f4071e
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 4 deletions.
diff --git a/src/teamster/core/google/sheets/assets.py b/src/teamster/core/google/sheets/assets.py
@@ -1,6 +1,6 @@
 import re
-from datetime import datetime
 
+import pendulum
 from dagster import DataVersion, observable_source_asset
 
 
@@ -15,6 +15,6 @@ def build_gsheet_asset(code_location, source_name, name, uri, range_name):
         metadata={"sheet_id": re_match.group(1), "range_name": range_name},
     )
     def _asset():
-        return DataVersion(str(datetime.now().timestamp()))
+        return DataVersion(str(pendulum.now().timestamp()))
 
     return _asset
diff --git a/src/teamster/core/utils/ops.py b/src/teamster/core/utils/ops.py
@@ -1,4 +1,6 @@
+import pendulum
 from dagster import AssetObservation, Config, OpExecutionContext, op
+from dagster._core.definitions.data_version import DATA_VERSION_TAG
 
 
 class ObservationOpConfig(Config):
@@ -8,4 +10,9 @@ class ObservationOpConfig(Config):
 @op
 def asset_observation_op(context: OpExecutionContext, config: ObservationOpConfig):
     for asset_key in config.asset_keys:
-        context.log_event(AssetObservation(asset_key=asset_key))
+        context.log_event(
+            AssetObservation(
+                asset_key=asset_key,
+                tags={DATA_VERSION_TAG: str(pendulum.now().timestamp())},
+            )
+        )
diff --git a/tests/adp/test_resource_adp.py b/tests/adp/test_resource_adp.py
@@ -25,10 +25,18 @@ def test_event_notification():
 
 
 def test_get_worker():
-    aoid = "G3R8E9HV8QXWF7JD"
+    aoid = "G3MQ5XDMH0DC9TWJ"
 
     r = ADP_WFN._request(
         method="GET", url=f"{ADP_WFN._service_root}/hr/v2/workers/{aoid}"
     )
 
     print(r.json())
+
+
+def test_get_workers():
+    r = ADP_WFN._request(method="GET", url=f"{ADP_WFN._service_root}/hr/v2/workers")
+
+    workers = r.json()["workers"]
+
+    print([w for w in workers if w["associateOID"] == "G3MQ5XDMH0DC9TWJ"])