filters all the way through to gql layer

python_modules/dagster-graphql/dagster_graphql/implementation/fetch_backfills.py

@@ -1,6 +1,6 @@
 from typing import TYPE_CHECKING, Optional
 
-from dagster._core.execution.backfill import BulkActionStatus
+from dagster._core.execution.backfill import BulkActionsFilter, BulkActionStatus
 
 if TYPE_CHECKING:
     from dagster_graphql.schema.util import ResolveInfo
@@ -23,11 +23,12 @@ def get_backfills(
     status: Optional[BulkActionStatus] = None,
     cursor: Optional[str] = None,
     limit: Optional[int] = None,
+    filters: Optional[BulkActionsFilter] = None,
 ) -> "GraphenePartitionBackfills":
     from ..schema.backfill import GraphenePartitionBackfill, GraphenePartitionBackfills
 
     backfills = graphene_info.context.instance.get_backfills(
-        status=status, cursor=cursor, limit=limit
+        status=status, cursor=cursor, limit=limit, filters=filters
     )
     return GraphenePartitionBackfills(
         results=[GraphenePartitionBackfill(backfill) for backfill in backfills]

python_modules/dagster-graphql/dagster_graphql/implementation/fetch_runs.py

@@ -19,6 +19,7 @@
 )
 from dagster._core.definitions.selector import JobSubsetSelector
 from dagster._core.errors import DagsterInvariantViolationError, DagsterRunNotFoundError
+from dagster._core.execution.backfill import BulkActionsFilter
 from dagster._core.instance import DagsterInstance
 from dagster._core.storage.dagster_run import DagsterRunStatus, RunRecord, RunsFilter
 from dagster._core.storage.event_log.base import AssetRecord
@@ -490,7 +491,7 @@ def get_runs_feed_entries(
         for backfill in instance.get_backfills(
             cursor=runs_feed_cursor.backfill_cursor,
             limit=limit,
-            created_before=created_before_cursor,
+            filters=BulkActionsFilter(created_before=created_before_cursor),
         )
     ]
     runs = [

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

@@ -1,5 +1,6 @@
 import graphene
 from dagster._core.events import DagsterEventType
+from dagster._core.execution.backfill import BulkActionsFilter, BulkActionStatus
 from dagster._core.storage.dagster_run import DagsterRunStatus, RunsFilter
 from dagster._time import datetime_from_timestamp
 from dagster._utils import check
@@ -373,6 +374,33 @@ class Meta:
         name = "TagInput"
 
 
+class GrapheneBulkActionsFilter(graphene.InputObjectType):
+    status = graphene.List(
+        graphene.NonNull("dagster_graphql.schema.backfill.GrapheneBulkActionStatus")
+    )
+    createdBefore = graphene.InputField(graphene.Float)
+    createdAfter = graphene.InputField(graphene.Float)
+
+    class Meta:
+        description = """This type represents a filter on Dagster Bulk Actions (backfills)."""
+        name = "BulkActionsFilter"
+
+    def to_selector(self):
+        if self.status:
+            status = BulkActionStatus.from_graphql_input(self.status)
+        else:
+            status = None
+
+        created_before = datetime_from_timestamp(self.createdBefore) if self.createdBefore else None
+        created_after = datetime_from_timestamp(self.createdAfter) if self.createdAfter else None
+
+        return BulkActionsFilter(
+            status=status,
+            created_before=created_before,
+            created_after=created_after,
+        )
+
+
 types = [
     GrapheneAssetKeyInput,
     GrapheneExecutionMetadata,
@@ -394,4 +422,5 @@ class Meta:
     GrapheneStepOutputHandle,
     GrapheneTagInput,
     GrapheneReportRunlessAssetEventsParams,
+    GrapheneBulkActionsFilter,
 ]

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

@@ -127,6 +127,7 @@
     GrapheneAssetBackfillPreviewParams,
     GrapheneAssetGroupSelector,
     GrapheneAssetKeyInput,
+    GrapheneBulkActionsFilter,
     GrapheneGraphSelector,
     GrapheneInstigationSelector,
     GraphenePipelineSelector,
@@ -474,6 +475,7 @@ class Meta:
         status=graphene.Argument(GrapheneBulkActionStatus),
         cursor=graphene.String(),
         limit=graphene.Int(),
+        filters=graphene.Argument(GrapheneBulkActionsFilter),
         description="Retrieve backfills after applying a status filter, cursor, and limit.",
     )
 
@@ -1105,12 +1107,14 @@ def resolve_partitionBackfillsOrError(
         status: Optional[GrapheneBulkActionStatus] = None,
         cursor: Optional[str] = None,
         limit: Optional[int] = None,
+        filters: Optional[GrapheneBulkActionsFilter] = None,
     ):
         return get_backfills(
             graphene_info,
             status=BulkActionStatus.from_graphql_input(status) if status else None,
             cursor=cursor,
             limit=limit,
+            filters=filters.to_selector() if filters else None,
         )
 
     def resolve_assetBackfillPreview(

python_modules/dagster-graphql/dagster_graphql_tests/graphql/test_asset_backfill.py

@@ -129,6 +129,20 @@
 }
 """
 
+BACKFILLS_WITH_FILTERS_QUERY = """
+  query BackfillsWithFiltersQuery($filters: BulkActionsFilter) {
+    partitionBackfillsOrError(filters: $filters) {
+      ... on PartitionBackfills {
+        results {
+          id
+          timestamp
+          status
+        }
+      }
+    }
+  }
+"""
+
 
 def get_repo() -> RepositoryDefinition:
     partitions_def = StaticPartitionsDefinition(["a", "b", "c"])
@@ -1056,7 +1070,9 @@ def test_asset_backfill_error_raised_upon_invalid_params_provided():
 
 
 def _get_backfill_data(
-    launch_backfill_result: GqlResult, instance: DagsterInstance, repo
+    launch_backfill_result: GqlResult,
+    instance: DagsterInstance,
+    repo,
 ) -> Tuple[str, AssetBackfillData]:
     assert launch_backfill_result
     assert launch_backfill_result.data
@@ -1131,3 +1147,56 @@ def override_backfill_storage_setting(self):
             )
 
             assert len(backfill_logs.data["partitionBackfillOrError"]["logEvents"]["events"]) > 0
+
+
+def test_get_backfills_with_filters():
+    repo = get_repo()
+    all_asset_keys = repo.asset_graph.materializable_asset_keys
+
+    with instance_for_test() as instance:
+        with define_out_of_process_context(__file__, "get_repo", instance) as context:
+            # launchPartitionBackfill
+            all_backfills = []
+            for _ in range(5):
+                launch_backfill_result = execute_dagster_graphql(
+                    context,
+                    LAUNCH_PARTITION_BACKFILL_MUTATION,
+                    variables={
+                        "backfillParams": {
+                            "partitionNames": ["a", "b"],
+                            "assetSelection": [key.to_graphql_input() for key in all_asset_keys],
+                        }
+                    },
+                )
+                assert (
+                    "backfillId" in launch_backfill_result.data["launchPartitionBackfill"]
+                ), _get_error_message(launch_backfill_result)
+
+                backfill_id = launch_backfill_result.data["launchPartitionBackfill"]["backfillId"]
+                backfill = instance.get_backfill(backfill_id)
+                all_backfills.append(backfill)
+
+            # on PartitionBackfills
+            get_backfills_result = execute_dagster_graphql(
+                context,
+                BACKFILLS_WITH_FILTERS_QUERY,
+                variables={"filters": {"createdBefore": all_backfills[3].backfill_timestamp}},
+            )
+            assert not get_backfills_result.errors
+            assert get_backfills_result.data
+            backfill_results = get_backfills_result.data["partitionBackfillsOrError"]["results"]
+
+            for result in backfill_results:
+                assert result["timestamp"] < all_backfills[3].backfill_timestamp
+
+            get_backfills_result = execute_dagster_graphql(
+                context,
+                BACKFILLS_WITH_FILTERS_QUERY,
+                variables={"filters": {"createdAfter": all_backfills[3].backfill_timestamp}},
+            )
+            assert not get_backfills_result.errors
+            assert get_backfills_result.data
+            backfill_results = get_backfills_result.data["partitionBackfillsOrError"]["results"]
+
+            for result in backfill_results:
+                assert result["timestamp"] > all_backfills[3].backfill_timestamp

python_modules/dagster/dagster/_core/execution/backfill.py

@@ -1,3 +1,4 @@
+from datetime import datetime
 from enum import Enum
 from typing import Mapping, NamedTuple, Optional, Sequence, Union
 
@@ -14,6 +15,7 @@
 from dagster._core.remote_representation.origin import RemotePartitionSetOrigin
 from dagster._core.storage.tags import USER_TAG
 from dagster._core.workspace.workspace import IWorkspace
+from dagster._record import record
 from dagster._serdes import whitelist_for_serdes
 from dagster._utils.error import SerializableErrorInfo
 
@@ -38,6 +40,13 @@ def from_graphql_input(graphql_str):
         return BulkActionStatus(graphql_str)
 
 
+@record
+class BulkActionsFilter:
+    status: Optional[BulkActionStatus] = None
+    created_before: Optional[datetime] = None
+    created_after: Optional[datetime] = None
+
+
 @whitelist_for_serdes
 class PartitionBackfill(
     NamedTuple(

python_modules/dagster/dagster/_core/instance/__init__.py

@@ -6,7 +6,6 @@
 import weakref
 from abc import abstractmethod
 from collections import defaultdict
-from datetime import datetime
 from enum import Enum
 from tempfile import TemporaryDirectory
 from types import TracebackType
@@ -124,7 +123,11 @@
         JobFailureData,
     )
     from dagster._core.events.log import EventLogEntry
-    from dagster._core.execution.backfill import BulkActionStatus, PartitionBackfill
+    from dagster._core.execution.backfill import (
+        BulkActionsFilter,
+        BulkActionStatus,
+        PartitionBackfill,
+    )
     from dagster._core.execution.plan.plan import ExecutionPlan
     from dagster._core.execution.plan.resume_retry import ReexecutionStrategy
     from dagster._core.execution.stats import RunStepKeyStatsSnapshot
@@ -3077,15 +3080,10 @@ def get_backfills(
         status: Optional["BulkActionStatus"] = None,
         cursor: Optional[str] = None,
         limit: Optional[int] = None,
-        created_before: Optional[datetime] = None,
-        created_after: Optional[datetime] = None,
+        filters: Optional["BulkActionsFilter"] = None,
     ) -> Sequence["PartitionBackfill"]:
         return self._run_storage.get_backfills(
-            status=status,
-            cursor=cursor,
-            limit=limit,
-            created_before=created_before,
-            created_after=created_after,
+            status=status, cursor=cursor, limit=limit, filters=filters
         )
 
     def get_backfill(self, backfill_id: str) -> Optional["PartitionBackfill"]:

python_modules/dagster/dagster/_core/storage/legacy_storage.py

@@ -1,4 +1,3 @@
-from datetime import datetime
 from typing import TYPE_CHECKING, Iterable, Mapping, Optional, Sequence, Set, Tuple, Union
 
 from dagster import _check as check
@@ -34,7 +33,11 @@
     from dagster._core.event_api import AssetRecordsFilter, RunStatusChangeRecordsFilter
     from dagster._core.events import DagsterEvent, DagsterEventType
     from dagster._core.events.log import EventLogEntry
-    from dagster._core.execution.backfill import BulkActionStatus, PartitionBackfill
+    from dagster._core.execution.backfill import (
+        BulkActionsFilter,
+        BulkActionStatus,
+        PartitionBackfill,
+    )
     from dagster._core.execution.stats import RunStepKeyStatsSnapshot
     from dagster._core.instance import DagsterInstance
     from dagster._core.remote_representation.origin import RemoteJobOrigin
@@ -313,12 +316,9 @@ def get_backfills(
         status: Optional["BulkActionStatus"] = None,
         cursor: Optional[str] = None,
         limit: Optional[int] = None,
-        created_before: Optional[datetime] = None,
-        created_after: Optional[datetime] = None,
+        filters: Optional["BulkActionsFilter"] = None,
     ) -> Sequence["PartitionBackfill"]:
-        return self._storage.run_storage.get_backfills(
-            status, cursor, limit, created_before=created_before, created_after=created_after
-        )
+        return self._storage.run_storage.get_backfills(status, cursor, limit, filters=filters)
 
     def get_backfill(self, backfill_id: str) -> Optional["PartitionBackfill"]:
         return self._storage.run_storage.get_backfill(backfill_id)