feat(reports): per-chart dashboard report capture behind feature flag (POC)

Add PER_CHART_DASHBOARD_REPORTS feature flag (default off). When enabled,
dashboard reports capture each chart individually and stack the images in
a single column, instead of one monolithic full-dashboard screenshot.

The report scheduler navigates to the dashboard permalink once — so tab
state and dashboard filters apply to every chart — then iterates over each
chart holder, waiting only for that chart's own loading spinner before
capturing it. One slow chart can no longer block or blank the entire
report; charts that never finish loading are skipped with a warning while
the rest are still delivered.

This mirrors the per-component delivery model used by Metabase (per-card
subscriptions) and Looker ("arrange dashboard tiles in a single column").

- take_per_chart_screenshots() in screenshot_utils.py: per-chart iteration,
  scroll-into-view for lazy loading, per-chart spinner wait, skip-on-timeout
- WebDriverPlaywright.get_per_chart_screenshots(): single navigation with
  auth and permalink context, Playwright-only
- DashboardScreenshot.get_per_chart_screenshots(): returns None on Selenium
  so callers fall back to the full-dashboard screenshot
- _get_screenshots() in execute.py: uses per-chart capture for dashboard
  reports when the flag is on; falls back to the full screenshot if no
  charts were captured
- Existing build_pdf_from_screenshots() stacks the per-chart images as PDF
  pages; PNG/Slack formats already accept image lists

POC for sc-113551. Not yet included: per-report-schedule toggle in the UI,
chart title/header treatment, placeholder for skipped charts.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

superset/commands/report/execute.py

@@ -463,9 +463,22 @@ class BaseReportState:
                 )
                 for url in urls
             ]
+        per_chart_reports = feature_flag_manager.is_feature_enabled(
+            "PER_CHART_DASHBOARD_REPORTS"
+        )
         try:
             imges = []
             for screenshot in screenshots:
+                if per_chart_reports and isinstance(screenshot, DashboardScreenshot):
+                    chart_imges = screenshot.get_per_chart_screenshots(user=user)
+                    if chart_imges:
+                        imges.extend(chart_imges)
+                        continue
+                    logger.warning(
+                        "Per-chart capture returned no charts for %s; "
+                        "falling back to full-dashboard screenshot",
+                        screenshot.url,
+                    )
                 imge = screenshot.get_screenshot(user=user)
                 if imge is None:
                     raise ReportScheduleScreenshotFailedError(

superset/config.py

@@ -732,6 +732,12 @@ DEFAULT_FEATURE_FLAGS: dict[str, bool] = {
     # When impersonating a user, use the email prefix instead of username
     # @lifecycle: testing
     "IMPERSONATE_WITH_EMAIL_PREFIX": False,
+    # Capture dashboard reports chart-by-chart and arrange the images in a
+    # single column, instead of one full-dashboard screenshot. Each chart
+    # waits for its own loading spinner, so one slow chart cannot block or
+    # blank the entire report. Requires PLAYWRIGHT_REPORTS_AND_THUMBNAILS.
+    # @lifecycle: testing
+    "PER_CHART_DASHBOARD_REPORTS": False,
     # Replace Selenium with Playwright for reports and thumbnails.
     # Supports deck.gl visualizations. Requires playwright pip package.
     # @lifecycle: testing

superset/utils/screenshot_utils.py

@@ -232,3 +232,69 @@ def take_tiled_screenshot(
     except Exception as e:
         logger.exception("Tiled screenshot failed: %s", e)
         return None
+
+
+CHART_HOLDER_SELECTOR = '[data-test="dashboard-component-chart-holder"]'
+
+
+def take_per_chart_screenshots(
+    page: "Page",
+    load_wait: int = 60,
+    animation_wait: int = 0,
+) -> list[bytes]:
+    """
+    Capture each chart on a dashboard as an individual screenshot.
+
+    Unlike the full-dashboard screenshot, each chart waits only for its own
+    loading spinner, so one slow chart cannot block or blank the entire
+    report. Charts whose spinner never clears within ``load_wait`` are
+    skipped with a warning rather than captured mid-load.
+
+    The page must already be navigated to the dashboard (typically via a
+    permalink URL so tab state and dashboard filters are applied).
+
+    Args:
+        page: Playwright page object, already at the dashboard URL
+        load_wait: Seconds to wait for each chart's spinner to clear
+        animation_wait: Seconds to wait for chart animations after load
+
+    Returns:
+        List of PNG screenshot bytes, one per successfully captured chart,
+        in DOM (layout) order. Empty list if no charts could be captured.
+    """
+    chart_holders = page.locator(CHART_HOLDER_SELECTOR)
+    chart_count = chart_holders.count()
+    logger.info("Capturing %s charts individually", chart_count)
+
+    screenshots: list[bytes] = []
+    for i in range(chart_count):
+        holder = chart_holders.nth(i)
+        try:
+            holder.scroll_into_view_if_needed()
+            # Trigger lazy loading and let layout settle
+            page.wait_for_timeout(SCROLL_SETTLE_TIMEOUT_MS)
+            try:
+                # Wait for this chart's own spinner only; spinners in other
+                # charts (still loading elsewhere on the page) don't block.
+                holder.locator(".loading").first.wait_for(
+                    state="detached", timeout=load_wait * 1000
+                )
+            except PlaywrightTimeout:
+                logger.warning(
+                    "Chart %s/%s did not finish loading within %ss; skipping",
+                    i + 1,
+                    chart_count,
+                    load_wait,
+                )
+                continue
+            if animation_wait > 0:
+                page.wait_for_timeout(animation_wait * 1000)
+            screenshots.append(holder.screenshot())
+            logger.debug("Captured chart %s/%s", i + 1, chart_count)
+        except Exception:
+            logger.exception(
+                "Failed to capture chart %s/%s; skipping", i + 1, chart_count
+            )
+
+    logger.info("Captured %s/%s charts individually", len(screenshots), chart_count)
+    return screenshots

superset/utils/screenshots.py

@@ -410,6 +410,24 @@ class DashboardScreenshot(BaseScreenshot):
         self.window_size = window_size or DEFAULT_DASHBOARD_WINDOW_SIZE
         self.thumb_size = thumb_size or DEFAULT_DASHBOARD_THUMBNAIL_SIZE
 
+    def get_per_chart_screenshots(
+        self, user: User, window_size: WindowSize | None = None
+    ) -> list[bytes] | None:
+        """
+        Capture each chart on the dashboard as an individual screenshot.
+
+        Playwright-only; returns None when the resolved driver is Selenium
+        so callers can fall back to the full-dashboard screenshot.
+        """
+        driver = self.driver(window_size, user)
+        if not isinstance(driver, WebDriverPlaywright):
+            logger.info(
+                "Per-chart screenshots require Playwright; falling back to "
+                "full-dashboard screenshot"
+            )
+            return None
+        return driver.get_per_chart_screenshots(self.url, self.element, user)
+
     def get_cache_key(
         self,
         window_size: bool | WindowSize | None = None,

superset/utils/webdriver.py

@@ -41,7 +41,10 @@ from selenium.webdriver.support.ui import WebDriverWait
 
 from superset.extensions import machine_auth_provider_factory
 from superset.utils.retries import retry_call
-from superset.utils.screenshot_utils import take_tiled_screenshot
+from superset.utils.screenshot_utils import (
+    take_per_chart_screenshots,
+    take_tiled_screenshot,
+)
 
 WindowSize = tuple[int, int]
 logger = logging.getLogger(__name__)
@@ -481,6 +484,78 @@ class WebDriverPlaywright(WebDriverProxy):
             context.close()
         return img
 
+    def get_per_chart_screenshots(
+        self, url: str, element_name: str, user: User | None = None
+    ) -> list[bytes] | None:
+        """
+        Capture each chart on a dashboard as an individual screenshot.
+
+        Navigates to the dashboard once (so permalink tab state and dashboard
+        filters apply to every chart), then captures chart-by-chart. Each
+        chart waits only for its own loading spinner, so one slow chart
+        cannot block or blank the entire report.
+
+        Returns a list of PNG bytes in layout order, or None if Playwright
+        is unavailable or the dashboard itself failed to load.
+        """
+        if not PLAYWRIGHT_AVAILABLE:
+            logger.info(
+                "Playwright not available - per-chart screenshots require "
+                "Playwright. %s",
+                PLAYWRIGHT_INSTALL_MESSAGE,
+            )
+            return None
+
+        browser_args = app.config["WEBDRIVER_OPTION_ARGS"]
+        browser = _browser_manager.get_browser(browser_args)
+        pixel_density = app.config["WEBDRIVER_WINDOW"].get("pixel_density", 1)
+        context = browser.new_context(
+            bypass_csp=True,
+            viewport={
+                "height": self._window[1],
+                "width": self._window[0],
+            },
+            device_scale_factor=pixel_density,
+        )
+        context.set_default_timeout(app.config["SCREENSHOT_PLAYWRIGHT_DEFAULT_TIMEOUT"])
+        if user:
+            self.auth(user, context)
+        page = context.new_page()
+        try:
+            try:
+                page.goto(
+                    url,
+                    wait_until=app.config["SCREENSHOT_PLAYWRIGHT_WAIT_EVENT"],
+                )
+            except PlaywrightTimeout:
+                logger.exception(
+                    "Web event %s not detected. Page %s might not have been "
+                    "fully loaded",
+                    app.config["SCREENSHOT_PLAYWRIGHT_WAIT_EVENT"],
+                    url,
+                )
+
+            selenium_headstart = app.config["SCREENSHOT_SELENIUM_HEADSTART"]
+            page.wait_for_timeout(selenium_headstart * 1000)
+            try:
+                page.locator(f".{element_name}").wait_for()
+            except PlaywrightTimeout:
+                logger.exception("Timed out requesting url %s", url)
+                return None
+
+            return take_per_chart_screenshots(
+                page,
+                load_wait=self._screenshot_load_wait,
+                animation_wait=app.config["SCREENSHOT_SELENIUM_ANIMATION_WAIT"],
+            )
+        except PlaywrightError:
+            logger.exception(
+                "Encountered an unexpected error when requesting url %s", url
+            )
+            return None
+        finally:
+            context.close()
+
 
 class WebDriverSelenium(WebDriverProxy):
     def __init__(

tests/unit_tests/commands/report/execute_test.py

@@ -54,7 +54,7 @@ from superset.reports.models import (
     ReportState,
 )
 from superset.utils.core import HeaderDataType
-from superset.utils.screenshots import ChartScreenshot
+from superset.utils.screenshots import ChartScreenshot, DashboardScreenshot
 from tests.integration_tests.conftest import with_feature_flags
 
 
@@ -1722,3 +1722,85 @@ def test_get_url_for_csv_uses_post_processed_type(
         f"CSV report URL must use type=post_processed so chart filters "
         f"(incl. time filters) are applied; got: {url}; see issue #25538"
     )
+
+
+def _make_dashboard_report_state(mocker: MockerFixture) -> BaseReportState:
+    """Build a BaseReportState for a dashboard report with screenshot mocks."""
+    mock_report_schedule: ReportSchedule = mocker.Mock(spec=ReportSchedule)
+    mock_report_schedule.chart = False
+    mock_report_schedule.dashboard_id = 123
+    mock_report_schedule.dashboard.digest = "digest"
+    mock_report_schedule.custom_width = None
+    mock_report_schedule.custom_height = None
+
+    mocker.patch(
+        "superset.commands.report.execute.get_executor",
+        return_value=(None, "admin"),
+    )
+    mocker.patch(
+        "superset.commands.report.execute.security_manager.find_user",
+        return_value=mocker.Mock(),
+    )
+    mocker.patch.object(
+        BaseReportState,
+        "get_dashboard_urls",
+        return_value=["http://example.com/superset/dashboard/p/abc123/"],
+    )
+
+    class_instance = BaseReportState(
+        mock_report_schedule, "January 1, 2021", "execution_id_example"
+    )
+    class_instance._report_schedule = mock_report_schedule
+    return class_instance
+
+
+@with_feature_flags(PER_CHART_DASHBOARD_REPORTS=True)
+def test_get_screenshots_per_chart_mode(mocker: MockerFixture, app) -> None:
+    """With the flag on, dashboard reports capture each chart individually."""
+    class_instance = _make_dashboard_report_state(mocker)
+    per_chart = mocker.patch.object(
+        DashboardScreenshot,
+        "get_per_chart_screenshots",
+        return_value=[b"chart1", b"chart2"],
+    )
+    full_screenshot = mocker.patch.object(DashboardScreenshot, "get_screenshot")
+
+    result = class_instance._get_screenshots()
+
+    assert result == [b"chart1", b"chart2"]
+    per_chart.assert_called_once()
+    full_screenshot.assert_not_called()
+
+
+@with_feature_flags(PER_CHART_DASHBOARD_REPORTS=True)
+def test_get_screenshots_per_chart_mode_falls_back_when_empty(
+    mocker: MockerFixture, app
+) -> None:
+    """If per-chart capture yields nothing, fall back to the full screenshot."""
+    class_instance = _make_dashboard_report_state(mocker)
+    mocker.patch.object(
+        DashboardScreenshot, "get_per_chart_screenshots", return_value=[]
+    )
+    full_screenshot = mocker.patch.object(
+        DashboardScreenshot, "get_screenshot", return_value=b"full_dashboard"
+    )
+
+    result = class_instance._get_screenshots()
+
+    assert result == [b"full_dashboard"]
+    full_screenshot.assert_called_once()
+
+
+@with_feature_flags(PER_CHART_DASHBOARD_REPORTS=False)
+def test_get_screenshots_per_chart_mode_disabled(mocker: MockerFixture, app) -> None:
+    """With the flag off, dashboard reports use the full-dashboard screenshot."""
+    class_instance = _make_dashboard_report_state(mocker)
+    per_chart = mocker.patch.object(DashboardScreenshot, "get_per_chart_screenshots")
+    mocker.patch.object(
+        DashboardScreenshot, "get_screenshot", return_value=b"full_dashboard"
+    )
+
+    result = class_instance._get_screenshots()
+
+    assert result == [b"full_dashboard"]
+    per_chart.assert_not_called()

tests/unit_tests/utils/test_screenshot_utils.py

@@ -23,7 +23,9 @@ from PIL import Image
 
 from superset.utils.screenshot_utils import (
     combine_screenshot_tiles,
+    PlaywrightTimeout,
     SCROLL_SETTLE_TIMEOUT_MS,
+    take_per_chart_screenshots,
     take_tiled_screenshot,
 )
 
@@ -397,3 +399,90 @@ class TestTakeTiledScreenshot:
 
         sig = inspect.signature(take_tiled_screenshot)
         assert sig.parameters["animation_wait"].default == 0
+
+
+class TestTakePerChartScreenshots:
+    def _make_page_with_charts(self, chart_screenshots: list[bytes]):
+        """Create a mock page whose chart holders return the given bytes."""
+        page = MagicMock()
+        holders = MagicMock()
+        holders.count.return_value = len(chart_screenshots)
+
+        holder_mocks = []
+        for shot in chart_screenshots:
+            holder = MagicMock()
+            holder.screenshot.return_value = shot
+            holder_mocks.append(holder)
+        holders.nth.side_effect = lambda i: holder_mocks[i]
+        page.locator.return_value = holders
+        return page, holder_mocks
+
+    def test_captures_all_charts_in_order(self):
+        page, _ = self._make_page_with_charts([b"chart1", b"chart2", b"chart3"])
+
+        result = take_per_chart_screenshots(page, load_wait=10)
+
+        assert result == [b"chart1", b"chart2", b"chart3"]
+
+    def test_uses_chart_holder_selector(self):
+        page, _ = self._make_page_with_charts([b"chart1"])
+
+        take_per_chart_screenshots(page, load_wait=10)
+
+        page.locator.assert_called_once_with(
+            '[data-test="dashboard-component-chart-holder"]'
+        )
+
+    def test_no_charts_returns_empty_list(self):
+        page, _ = self._make_page_with_charts([])
+
+        result = take_per_chart_screenshots(page, load_wait=10)
+
+        assert result == []
+
+    def test_spinner_timeout_skips_chart_and_continues(self):
+        page, holders = self._make_page_with_charts([b"chart1", b"chart2"])
+        # First chart's spinner never clears
+        holders[0].locator.return_value.first.wait_for.side_effect = PlaywrightTimeout(
+            "spinner stuck"
+        )
+
+        result = take_per_chart_screenshots(page, load_wait=10)
+
+        assert result == [b"chart2"]
+        holders[0].screenshot.assert_not_called()
+
+    def test_screenshot_failure_skips_chart_and_continues(self):
+        page, holders = self._make_page_with_charts([b"chart1", b"chart2"])
+        holders[0].screenshot.side_effect = Exception("capture failed")
+
+        result = take_per_chart_screenshots(page, load_wait=10)
+
+        assert result == [b"chart2"]
+
+    def test_waits_for_each_charts_own_spinner(self):
+        page, holders = self._make_page_with_charts([b"chart1"])
+
+        take_per_chart_screenshots(page, load_wait=42)
+
+        holders[0].locator.assert_called_once_with(".loading")
+        holders[0].locator.return_value.first.wait_for.assert_called_once_with(
+            state="detached", timeout=42000
+        )
+
+    def test_scrolls_each_chart_into_view(self):
+        page, holders = self._make_page_with_charts([b"chart1", b"chart2"])
+
+        take_per_chart_screenshots(page, load_wait=10)
+
+        for holder in holders:
+            holder.scroll_into_view_if_needed.assert_called_once()
+
+    def test_animation_wait_applied_after_spinner_clears(self):
+        page, _ = self._make_page_with_charts([b"chart1"])
+
+        take_per_chart_screenshots(page, load_wait=10, animation_wait=3)
+
+        # Scroll settle wait plus animation wait
+        page.wait_for_timeout.assert_any_call(SCROLL_SETTLE_TIMEOUT_MS)
+        page.wait_for_timeout.assert_any_call(3000)