add test and clean up types

js_modules/dagster-ui/packages/ui-core/src/app/Permissions.tsx

@@ -21,6 +21,7 @@ export const EXPECTED_PERMISSIONS = {
   reload_repository_location: true,
   reload_workspace: true,
   wipe_assets: true,
+  report_runless_asset_events: true,
   launch_partition_backfill: true,
   cancel_partition_backfill: true,
   edit_dynamic_partitions: true,

python_modules/dagster-graphql/dagster_graphql/implementation/execution/__init__.py

@@ -8,10 +8,10 @@
 from dagster._annotations import deprecated
 from dagster._core.definitions.events import AssetKey
 from dagster._core.events import (
-    EngineEventData,
-    DagsterEventType,
     AssetMaterialization,
     AssetObservation,
+    DagsterEventType,
+    EngineEventData,
 )
 from dagster._core.instance import (
     DagsterInstance,
@@ -51,8 +51,8 @@
     from ...schema.errors import GrapheneRunNotFoundError
     from ...schema.roots.mutation import (
         GrapheneAssetWipeSuccess,
-        GrapheneReportRunlessAssetEventsSuccess,
         GrapheneDeletePipelineRunSuccess,
+        GrapheneReportRunlessAssetEventsSuccess,
         GrapheneTerminateRunFailure,
         GrapheneTerminateRunsResult,
         GrapheneTerminateRunSuccess,
@@ -402,7 +402,9 @@ def report_runless_asset_events(
     asset_key: AssetKey,
     partition_keys: Optional[Sequence[str]] = None,
     description: Optional[str] = None,
-) -> "GrapheneAssetWipeSuccess":
+) -> "GrapheneReportRunlessAssetEventsSuccess":
+    from ...schema.roots.mutation import GrapheneReportRunlessAssetEventsSuccess
+
     instance = graphene_info.context.instance
 
     if partition_keys is not None:
@@ -415,4 +417,4 @@ def report_runless_asset_events(
             create_asset_event(event_type, asset_key, None, description)
         )
 
-    return GrapheneReportRunlessAssetEventsSuccess(asset_key=asset_key)
+    return GrapheneReportRunlessAssetEventsSuccess(assetKey=asset_key)

python_modules/dagster-graphql/dagster_graphql/schema/inputs.py

@@ -1,7 +1,8 @@
 import graphene
 import pendulum
-from dagster._core.storage.dagster_run import DagsterRunStatus, RunsFilter
 from dagster._core.events import DagsterEventType
+from dagster._core.storage.dagster_run import DagsterRunStatus, RunsFilter
+from dagster._utils import check
 
 from .pipelines.status import GrapheneRunStatus
 from .runs import GrapheneRunConfigData
@@ -189,7 +190,7 @@ class Meta:
         name = "LaunchBackfillParams"
 
 
-class GrapheneAssetEventType(graphene.Enum):
+class GrapheneRunlessAssetEventType(graphene.Enum):
     """The event type of an asset event."""
 
     ASSET_MATERIALIZATION = "ASSET_MATERIALIZATION"
@@ -199,18 +200,18 @@ class Meta:
         name = "AssetEventType"
 
     def to_dagster_event_type(self):
-        if self == GrapheneAssetEventType.ASSET_MATERIALIZATION:
+        if self == GrapheneRunlessAssetEventType.ASSET_MATERIALIZATION:
             return DagsterEventType.ASSET_MATERIALIZATION
-        elif self == GrapheneAssetEventType.ASSET_OBSERVATION:
+        elif self == GrapheneRunlessAssetEventType.ASSET_OBSERVATION:
             return DagsterEventType.ASSET_OBSERVATION
         else:
             check.assert_never(self)
 
 
 class GrapheneReportRunlessAssetEventsParams(graphene.InputObjectType):
-    eventType = graphene.NonNull(GrapheneAssetEventType)
+    eventType = graphene.NonNull(GrapheneRunlessAssetEventType)
     assetKey = graphene.NonNull(GrapheneAssetKeyInput)
-    partitionNames = graphene.InputField(graphene.List(graphene.String))
+    partitionKeys = graphene.InputField(graphene.List(graphene.String))
     description = graphene.String()
 
     class Meta:

python_modules/dagster-graphql/dagster_graphql/schema/roots/mutation.py

@@ -22,10 +22,10 @@
 
 from ...implementation.execution import (
     delete_pipeline_run,
+    report_runless_asset_events,
     terminate_pipeline_execution,
     terminate_pipeline_execution_for_runs,
     wipe_assets,
-    report_runless_asset_events,
 )
 from ...implementation.external import fetch_workspace, get_full_external_job_or_raise
 from ...implementation.telemetry import log_ui_telemetry_event
@@ -59,10 +59,10 @@
 from ..external import GrapheneWorkspace, GrapheneWorkspaceLocationEntry
 from ..inputs import (
     GrapheneAssetKeyInput,
-    GrapheneReportRunlessAssetEventsParams,
     GrapheneExecutionParams,
     GrapheneLaunchBackfillParams,
     GrapheneReexecutionParams,
+    GrapheneReportRunlessAssetEventsParams,
     GrapheneRepositorySelector,
 )
 from ..partition_sets import GrapheneAddDynamicPartitionResult
@@ -678,7 +678,7 @@ def mutate(self, graphene_info: ResolveInfo, assetKeys: Sequence[GrapheneAssetKe
 class GrapheneReportRunlessAssetEventsSuccess(graphene.ObjectType):
     """Output indicating that runless asset events were reported."""
 
-    assetKey = GrapheneAssetKey
+    assetKey = graphene.NonNull(GrapheneAssetKey)
 
     class Meta:
         name = "ReportRunlessAssetEventsSuccess"
@@ -689,7 +689,6 @@ class GrapheneReportRunlessAssetEventsResult(graphene.Union):
 
     class Meta:
         types = (
-            GrapheneAssetNotFoundError,
             GrapheneUnauthorizedError,
             GraphenePythonError,
             GrapheneReportRunlessAssetEventsSuccess,
@@ -703,7 +702,7 @@ class GrapheneReportRunlessAssetEventsMutation(graphene.Mutation):
     Output = graphene.NonNull(GrapheneReportRunlessAssetEventsResult)
 
     class Arguments:
-        eventParams = graphene.Argument(GrapheneReportRunlessAssetEventsParams)
+        eventParams = graphene.Argument(graphene.NonNull(GrapheneReportRunlessAssetEventsParams))
 
     class Meta:
         name = "ReportRunlessAssetEventsMutation"
@@ -713,10 +712,18 @@ class Meta:
     def mutate(
         self, graphene_info: ResolveInfo, eventParams: GrapheneReportRunlessAssetEventsParams
     ):
-        event_type = eventParams.eventType.to_dagster_event_type()
+        event_type = eventParams["eventType"].to_dagster_event_type()
         asset_key = AssetKey.from_graphql_input(eventParams["assetKey"])
+        partition_keys = eventParams.get("partitionKeys", None)
+        description = eventParams.get("description", None)
 
-        return report_runless_asset_events(graphene_info, AssetKey.from_graphql_input(eventParams))
+        return report_runless_asset_events(
+            graphene_info,
+            event_type=event_type,
+            asset_key=asset_key,
+            partition_keys=partition_keys,
+            description=description,
+        )
 
 
 class GrapheneLogTelemetrySuccess(graphene.ObjectType):