[gql] latest event id resolver for Asset so that we can order by id

Author: jamiedemaria

python_modules/dagster-graphql/dagster_graphql/schema/pipelines/pipeline.py

@@ -21,6 +21,7 @@
     RunRecord,
     RunsFilter,
 )
+from dagster._core.storage.event_log.base import AssetRecord
 from dagster._core.storage.tags import REPOSITORY_LABEL_TAG, RUN_METRIC_TAGS, TagType, get_tag_type
 from dagster._core.workspace.permissions import Permissions
 from dagster._utils.tags import get_boolean_tag_value
@@ -247,6 +248,7 @@ class GrapheneAsset(graphene.ObjectType):
         cursor=graphene.String(),
     )
     definition = graphene.Field("dagster_graphql.schema.asset_graph.GrapheneAssetNode")
+    latestEventId = graphene.Field(graphene.Int)
 
     class Meta:
         name = "Asset"
@@ -389,6 +391,12 @@ def resolve_assetObservations(
             )
         ]
 
+    async def resolve_LatestEventId(self, graphene_info):
+        asset_record = await AssetRecord.gen(graphene_info.context, self.key)
+        if asset_record:
+            return asset_record.asset_entry.last_event_storage_id
+        return None
+
 
 class GrapheneEventConnection(graphene.ObjectType):
     class Meta:

python_modules/dagster/dagster/_core/storage/event_log/base.py

@@ -130,6 +130,12 @@ def last_observation(self) -> Optional["EventLogEntry"]:
             return None
         return self.last_observation_record.event_log_entry
 
+    @property
+    def last_observation_storage_id(self) -> Optional[int]:
+        if self.last_observation_record is None:
+            return None
+        return self.last_observation_record.storage_id
+
     @property
     def last_materialization_storage_id(self) -> Optional[int]:
         if self.last_materialization_record is None:
@@ -148,6 +154,18 @@ def last_failed_to_materialize_storage_id(self) -> Optional[int]:
             return None
         return self.last_failed_to_materialize_record.storage_id
 
+    @property
+    def last_event_storage_id(self) -> Optional[int]:
+        """Get the storage id of the latest event for this asset."""
+        event_ids = [
+            self.last_materialization_storage_id,
+            self.last_observation_storage_id,
+            self.last_failed_to_materialize_storage_id,
+            self.last_planned_materialization_storage_id,
+        ]
+        event_ids = [event_id for event_id in event_ids if event_id is not None]
+        return max(event_ids) if event_ids else None
+
 
 class AssetRecord(
     NamedTuple("_NamedTuple", [("storage_id", int), ("asset_entry", AssetEntry)]),