Make sure video S3 objects are not re-uploaded every time a failed webhook is retried.

Author: mbertrand

dg_projects/learning_resources/learning_resources/assets/youtube_shorts.py

@@ -15,13 +15,18 @@
     AssetExecutionContext,
     AssetIn,
     AssetKey,
+    AssetRecordsFilter,
+    Backoff,
     Config,
     DataVersion,
     DynamicPartitionsDefinition,
     Output,
+    RetryPolicy,
     asset,
 )
 from ol_orchestrate.lib.automation_policies import upstream_or_code_changes
+from s3fs import S3FileSystem
+from upath import UPath
 
 from learning_resources.resources.api_client_factory import ApiClientFactory
 from learning_resources.resources.youtube_client import YouTubeClientFactory
@@ -48,6 +53,14 @@ def _assign_s3_path(video_id: str, extension: str) -> str:
     return f"{video_id}/{video_id}.{extension}"
 
 
+def _build_full_s3_path(config: YouTubeShortsConfig, relative_path: str) -> str:
+    """Build full S3 path from config and relative path."""
+    bucket = config.bucket_name
+    prefix = config.s3_prefix
+    full_path = f"{bucket}/{prefix}/{relative_path}".replace("//", "/")
+    return f"s3://{full_path}"
+
+
 def _load_metadata_from_config(
     config: YouTubeShortsConfig,
 ) -> dict[str, Any] | None:
@@ -84,6 +97,159 @@ def _generate_video_version(video_data: dict[str, Any]) -> str:
     return hashlib.sha256(version_string.encode()).hexdigest()[:16]
 
 
+def _should_skip_asset_materialization(
+    context: AssetExecutionContext,
+    expected_version: str,
+    s3_path: str,
+) -> tuple[bool, dict[str, Any]]:
+    """
+    Check if an asset should skip materialization based on existing version and S3 file.
+
+
+    Args:
+        context: Dagster execution context with asset_key and partition_key
+        expected_version: The expected data version hash for this asset
+        s3_path: Full S3 path (s3://bucket/key) to verify file existence
+
+    Returns:
+        Tuple of (should_skip: bool, existing_metadata: dict) where
+        should_skip is True if the asset already exists with the correct
+        version AND the S3 file exists with content.
+    """
+    video_id = context.partition_key
+
+    # Check if asset already exists with correct version
+    existing_materializations = context.instance.fetch_materializations(
+        records_filter=AssetRecordsFilter(
+            asset_key=context.asset_key,
+            asset_partitions=[video_id],
+        ),
+        limit=1,
+    )
+
+    if not existing_materializations.records:
+        log.debug(
+            "No existing materialization found for %s partition %s",
+            context.asset_key,
+            video_id,
+        )
+        return (False, {})
+
+    existing_record = existing_materializations.records[0]
+    existing_version = existing_record.asset_materialization.tags.get(
+        "dagster/data_version"
+    )
+    existing_metadata = {
+        k: v.value for k, v in existing_record.asset_materialization.metadata.items()
+    }
+
+    if existing_version != expected_version:
+        log.info(
+            "Asset %s for video %s exists but version mismatch "
+            "(existing: %s, expected: %s) - re-materializing",
+            context.asset_key,
+            video_id,
+            existing_version,
+            expected_version,
+        )
+        return (False, {})
+
+    # Version matches - verify S3 file actually exists before skipping
+    try:
+        s3_fs = S3FileSystem()
+        s3_file = UPath(s3_path, **s3_fs.storage_options)
+
+        if not s3_file.exists():
+            log.warning(
+                "Asset %s for video %s has correct version but S3 file %s "
+                "does not exist - re-materializing",
+                context.asset_key,
+                video_id,
+                s3_path,
+            )
+            return (False, {})
+
+        # Check file size to ensure it's not empty
+        file_size = s3_file.stat().st_size
+        if file_size == 0:
+            log.warning(
+                "Asset %s for video %s has correct version but S3 file is empty "
+                "(0 bytes) - re-materializing",
+                context.asset_key,
+                video_id,
+            )
+            return (False, {})
+
+        log.info(
+            "Asset %s for video %s already exists with correct version %s "
+            "and valid S3 file (%d bytes) - skipping",
+            context.asset_key,
+            video_id,
+            expected_version,
+            file_size,
+        )
+    except Exception as e:  # noqa: BLE001
+        log.warning(
+            "Failed to verify S3 file existence for %s: %s - "
+            "re-materializing to be safe",
+            s3_path,
+            e,
+            exc_info=True,
+        )
+        return (False, {})
+    else:
+        return (True, existing_metadata)
+
+
+def _create_skip_output(
+    context: AssetExecutionContext,
+    version: str,
+    target_path: str,
+    file_extension: str,
+    existing_metadata: dict[str, Any],
+) -> Output[tuple]:
+    """
+    Create an Output for skipping asset materialization.
+
+    Creates a lightweight marker file that the IO manager will detect and skip
+    uploading, preserving the existing asset in S3 without re-processing or
+    re-uploading.
+
+    Args:
+        context: Dagster execution context
+        version: Data version hash
+        target_path: Target S3 path for the asset
+            (relative path like "video_id/video_id.mp4")
+        file_extension: File extension for the marker file
+            (e.g., "mp4", "jpg", "json")
+        existing_metadata: Metadata from the existing materialization to preserve
+
+    Returns:
+        Output with marker file and preserved metadata
+    """
+    video_id = context.partition_key
+
+    # Create an empty marker file - IO manager will detect .skip_ prefix and skip upload
+    marker_file = Path(f".skip_{video_id}.{file_extension}")
+    marker_file.touch()
+
+    log.info(
+        "Skipping %s for video %s - already exists with correct version %s",
+        context.asset_key,
+        video_id,
+        version,
+    )
+
+    return Output(
+        value=(marker_file, target_path),
+        data_version=DataVersion(version),
+        metadata={
+            **existing_metadata,
+            "status": "skipped_existing",
+        },
+    )
+
+
 def fetch_video_metadata(youtube_client, video_id):
     request = youtube_client.client.videos().list(
         part="snippet,contentDetails", id=video_id
@@ -149,6 +315,18 @@ def youtube_video_metadata(
         # Generate version for change tracking
         version = _generate_video_version(video_data)
 
+        target_path = _assign_s3_path(video_id, "json")
+        full_s3_path = _build_full_s3_path(config, target_path)
+
+        # Check if asset already exists with correct version - skip re-processing
+        should_skip, existing_metadata = _should_skip_asset_materialization(
+            context, version, full_s3_path
+        )
+        if should_skip:
+            return _create_skip_output(
+                context, version, target_path, "json", existing_metadata
+            )
+
         # Write metadata to local JSON file for io_manager to upload
         metadata_file = Path(f"{version}.metadata.json")
         with metadata_file.open("w", encoding="utf-8") as f:
@@ -206,6 +384,18 @@ def youtube_video_content(
     # Generate version for change tracking
     version = _generate_video_version(video_data or {"id": video_id})
 
+    target_path = _assign_s3_path(video_id, "mp4")
+    full_s3_path = _build_full_s3_path(config, target_path)
+
+    # Check if asset already exists with correct version - skip expensive download
+    should_skip, existing_metadata = _should_skip_asset_materialization(
+        context, version, full_s3_path
+    )
+    if should_skip:
+        return _create_skip_output(
+            context, version, target_path, "mp4", existing_metadata
+        )
+
     # Download video using yt-dlp to local file
     output_file = Path(f"{video_id}.mp4")
 
@@ -317,6 +507,18 @@ def youtube_video_thumbnail(
     # Generate version for change tracking
     version = _generate_video_version(video_metadata)
 
+    target_path = _assign_s3_path(video_id, "jpg")
+    full_s3_path = _build_full_s3_path(config, target_path)
+
+    # Check if asset already exists with correct version - skip expensive download
+    should_skip, existing_metadata = _should_skip_asset_materialization(
+        context, version, full_s3_path
+    )
+    if should_skip:
+        return _create_skip_output(
+            context, version, target_path, "jpg", existing_metadata
+        )
+
     try:
         # Download thumbnail
         response = requests.get(thumbnail_url, timeout=30)
@@ -360,27 +562,59 @@ def youtube_video_thumbnail(
     code_version="youtube_webhook_v1",
     partitions_def=youtube_video_ids,
     ins={"video_metadata": AssetIn(key=AssetKey(["youtube_shorts", "video_metadata"]))},
+    retry_policy=RetryPolicy(max_retries=3, delay=1, backoff=Backoff.EXPONENTIAL),
 )
 def youtube_webhook(
     context: AssetExecutionContext,
+    config: YouTubeShortsConfig,
     learn_api: ApiClientFactory,
+    youtube_client: YouTubeClientFactory,
     video_metadata,  # noqa: ARG001
 ) -> Output[bool]:
     """
     Send a webhook notification for a processed YouTube video.
 
     Args:
         context: Dagster execution context containing partition key (video ID)
+        config: Configuration containing cached metadata from sensor
         learn_api: MIT Learn API client factory
-        video_metadata: S3 path to metadata file
+        youtube_client: YouTube API client factory
+        video_metadata: Dependency to ensure metadata is processed first
 
     Returns:
         Boolean indicating success of the webhook notification
     """
     video_id = context.partition_key
 
     try:
-        webhook_response = learn_api.client.notify_shorts_processed(video_id)
+        # Try to get metadata from the sensor via run_config
+        metadata_dict = _load_metadata_from_config(config)
+
+        if metadata_dict:
+            log.info("Using cached metadata from sensor for webhook: %s", video_id)
+        else:
+            # Fallback: Fetch metadata from YouTube API
+            log.info("Fetching metadata from YouTube for %s", video_id)
+            try:
+                response = fetch_video_metadata(youtube_client, video_id)
+                videos = response.get("items", [])
+                if not videos:
+                    msg = f"Video {video_id} not found in YouTube API"
+                    raise ValueError(msg)  # noqa: TRY301
+                metadata_dict = videos[0]
+            except Exception as fetch_error:
+                log.exception(
+                    "Failed to fetch metadata from YouTube API for video %s",
+                    video_id,
+                )
+                msg = (
+                    f"No metadata found in run config and failed to fetch "
+                    f"from YouTube API for video {video_id}. "
+                    "Cannot send webhook without metadata."
+                )
+                raise ValueError(msg) from fetch_error
+
+        webhook_response = learn_api.client.notify_shorts_processed(metadata_dict)
         log.info(
             "Successfully sent webhook notification for video %s: %s",
             video_id,