Get rid of unnecessary functions and revert # of videos to process

Author: mbertrand

dg_projects/learning_resources/learning_resources/sensors/youtube_shorts.py

@@ -12,15 +12,13 @@
 from dagster import (
     AddDynamicPartitionsRequest,
     AssetKey,
-    DagsterEventType,
     DefaultSensorStatus,
-    EventRecordsFilter,
     RunRequest,
     SensorResult,
     sensor,
 )
 
-MAX_VIDEO_RESULTS = 1
+MAX_VIDEO_RESULTS = 16
 
 
 def fetch_youtube_shorts_api_data(
@@ -130,61 +128,12 @@ def fetch_and_log_video_metadata(youtube_client, youtube_config_provider):
     return current_video_ids, metadata_by_video_id
 
 
-def _check_asset_materialization(context, asset_key, partition_key):
-    """
-    Check if a specific asset partition has been successfully materialized.
-
-    Args:
-        context: Sensor evaluation context
-        asset_key: The asset key to check
-        partition_key: The partition key (video ID) to check
-
-    Returns:
-        Boolean indicating if the asset partition has been successfully materialized
-    """
-    # Check for successful materialization
-    materializations = context.instance.get_event_records(
-        event_records_filter=EventRecordsFilter(
-            asset_key=asset_key,
-            asset_partitions=[partition_key],
-            event_type=DagsterEventType.ASSET_MATERIALIZATION,
-        ),
-        limit=1,
-    )
-
-    return len(materializations) > 0
-
-
-def _is_video_fully_processed(context, video_id):
-    """
-    Check if all assets for a video have been successfully materialized.
-
-    Args:
-        context: Sensor evaluation context
-        video_id: Video ID to check
-
-    Returns:
-        Boolean indicating if all assets have been materialized
-    """
-    required_assets = [
-        AssetKey(["youtube_shorts", "video_content"]),
-        AssetKey(["youtube_shorts", "video_thumbnail"]),
-        AssetKey(["youtube_shorts", "video_metadata"]),
-        AssetKey(["youtube_shorts", "video_webhook"]),
-    ]
-
-    for asset_key in required_assets:
-        if not _check_asset_materialization(context, asset_key, video_id):
-            return False
-
-    return True
-
-
 def create_run_requests(context, videos_to_process, metadata_by_video_id):
     """
     Create run requests for videos that need processing.
 
-    Only creates run requests for videos that haven't been fully processed yet.
+    Individual assets will determine if they need to be re-materialized based on
+    version and S3 file existence checks.
 
     Args:
         context: Sensor evaluation context
@@ -196,11 +145,6 @@ def create_run_requests(context, videos_to_process, metadata_by_video_id):
     """
     run_requests = []
     for video_id in videos_to_process:
-        # Skip videos that have already been fully processed
-        if _is_video_fully_processed(context, video_id):
-            context.log.debug("Video %s already fully processed, skipping", video_id)
-            continue
-
         context.log.info("Creating run request for video %s", video_id)
 
         metadata_json = (