cp for testing

python_modules/dagster/dagster/_core/asset_graph_view/cron.py

@@ -71,18 +71,26 @@ def get_missed_ticks(cron_data: CronEvaluationData) -> Sequence[datetime.datetim
     )
 
 
+# class NewPartitionRequest(NamedTuple):
+#     missed_ticks: Sequence[datetime.datetime]
+#     asset_partition_keys: AbstractSet[AssetKeyPartitionKey]
+
+
 def get_new_asset_partitions_to_request(
+    *,
     cron_data: CronEvaluationData,
     asset_key: AssetKey,
     partitions_def: Optional[PartitionsDefinition],
     dynamic_partitions_store: DynamicPartitionsStore,
     all_partitions: bool,
+    missed_ticks: Sequence[datetime.datetime],
 ) -> AbstractSet[AssetKeyPartitionKey]:
-    missed_ticks = (
-        cron_ticks_in_time_range(cron_data)
-        if cron_data.start_timestamp
-        else last_tick_in_cron_schedule(cron_data)
-    )
+    # missed_ticks = get_missed_ticks(cron_data)
+    # # (
+    #     cron_ticks_in_time_range(cron_data)
+    #     if cron_data.start_timestamp
+    #     else last_tick_in_cron_schedule(cron_data)
+    # )
 
     if not missed_ticks:
         return set()

python_modules/dagster/dagster/_core/definitions/auto_materialize_rule.py

@@ -8,7 +8,6 @@
     Mapping,
     NamedTuple,
     Optional,
-    Sequence,
     Set,
 )
 
@@ -291,50 +290,38 @@ def decision_type(self) -> AutoMaterializeDecisionType:
     def description(self) -> str:
         return f"not materialized since last cron schedule tick of '{self.cron_schedule}' (timezone: {self.timezone})"
 
-    def missed_cron_ticks(
-        self, context: AssetConditionEvaluationContext
-    ) -> Sequence[datetime.datetime]:
-        """Returns the cron ticks which have been missed since the previous cursor was generated."""
-        # if it's the first time evaluating this rule, then just count the latest tick as missed
-        start_dt = (
-            None
-            if (not context.previous_evaluation or not context.previous_evaluation_timestamp)
-            else datetime.datetime.fromtimestamp(context.previous_evaluation_timestamp)
-        )
-        return get_missed_ticks(
-            CronEvaluationData(
-                cron_schedule=self.cron_schedule,
-                cron_timezone=self.timezone,
-                start_dt=start_dt,
-                end_dt=context.evaluation_time,
-            )
-        )
-
     def evaluate_for_asset(
         self, context: AssetConditionEvaluationContext
     ) -> "AssetConditionResult":
         from .asset_condition.asset_condition import AssetConditionResult
 
         # if it's the first time evaluating this rule, then just count the latest tick as missed
-        first_time = not context.previous_evaluation or not context.previous_evaluation_timestamp
-        missed_ticks = get_missed_ticks(
-            CronEvaluationData(
-                cron_schedule=self.cron_schedule,
-                cron_timezone=self.timezone,
-                start_dt=None if first_time else context.evaluation_time,
-                end_dt=context.evaluation_time,
+        # by setting start_dt to None
+        first_tick = not context.previous_evaluation or not context.previous_evaluation_timestamp
+        start_dt = (
+            None
+            if first_tick
+            else datetime.datetime.fromtimestamp(
+                check.not_none(context.previous_evaluation_timestamp)
             )
         )
+        cron_data = CronEvaluationData(
+            cron_schedule=self.cron_schedule,
+            cron_timezone=self.timezone,
+            start_dt=start_dt,
+            end_dt=context.evaluation_time,
+        )
+
+        missed_ticks = get_missed_ticks(cron_data)
 
         new_asset_partitions = get_new_asset_partitions_to_request(
             cron_data=CronEvaluationData(
                 cron_schedule=self.cron_schedule,
                 cron_timezone=self.timezone,
-                start_dt=datetime.datetime.fromtimestamp(context.previous_evaluation_timestamp)
-                if context.previous_evaluation_timestamp
-                else None,
+                start_dt=start_dt,
                 end_dt=context.evaluation_time,
             ),
+            missed_ticks=missed_ticks,
             asset_key=context.asset_key,
             partitions_def=context.partitions_def,
             dynamic_partitions_store=context.instance_queryer,