From df7e2b6a8e5ce01cbc79d832e86fc4b8511f4385 Mon Sep 17 00:00:00 2001
From: Lily Kuang <lily@preset.io>
Date: Thu, 15 Apr 2021 14:07:49 -0700
Subject: [PATCH] feat(alert/report): chart as csv format attachment for email
 and slack (#13828)

* add ui for setting report format

* refactor default notification format

* init csv data alert report

* add report format to report_schedule model

* add ALERTS_ATTACH_REPORTS feature flag

* fix lint

* update check image tag

* fix migrations

Co-authored-by: samtfm <sam@preset.io>
---
 .../src/views/CRUD/alert/AlertReportModal.tsx | 211 +++--------
 .../alert/components/NotificationMethod.tsx   | 168 +++++++++
 .../src/views/CRUD/alert/types.ts             |   1 +
 ...3_add_report_format_to_report_schedule_.py |  44 +++
 superset/models/reports.py                    |   7 +-
 superset/reports/api.py                       |   4 +-
 superset/reports/commands/exceptions.py       |   8 +
 superset/reports/commands/execute.py          |  56 ++-
 superset/reports/notifications/base.py        |   3 +-
 superset/reports/notifications/email.py       |  15 +-
 superset/reports/notifications/slack.py       |  14 +-
 superset/reports/schemas.py                   |   9 +
 superset/utils/csv.py                         |  21 +-
 tests/reports/commands_tests.py               | 352 ++++++++++++++----
 tests/reports/utils.py                        |   3 +
 15 files changed, 670 insertions(+), 246 deletions(-)
 create mode 100644 superset-frontend/src/views/CRUD/alert/components/NotificationMethod.tsx
 create mode 100644 superset/migrations/versions/19e978e1b9c3_add_report_format_to_report_schedule_.py

diff --git a/superset-frontend/src/views/CRUD/alert/AlertReportModal.tsx b/superset-frontend/src/views/CRUD/alert/AlertReportModal.tsx
index ea9422cfa..6f4d57f52 100644
--- a/superset-frontend/src/views/CRUD/alert/AlertReportModal.tsx
+++ b/superset-frontend/src/views/CRUD/alert/AlertReportModal.tsx
@@ -26,10 +26,13 @@ import Modal from 'src/common/components/Modal';
 import { Switch } from 'src/common/components/Switch';
 import { Radio } from 'src/common/components/Radio';
 import { AsyncSelect, NativeGraySelect as Select } from 'src/components/Select';
+import { FeatureFlag, isFeatureEnabled } from 'src/featureFlags';
 import withToasts from 'src/messageToasts/enhancers/withToasts';
 import Owner from 'src/types/Owner';
 import TextAreaControl from 'src/explore/components/controls/TextAreaControl';
 import { AlertReportCronScheduler } from './components/AlertReportCronScheduler';
+import { NotificationMethod } from './components/NotificationMethod';
+
 import {
   AlertObject,
   ChartObject,
@@ -58,7 +61,7 @@ interface AlertReportModalProps {
 }
 
 const NOTIFICATION_METHODS: NotificationMethod[] = ['Email', 'Slack'];
-
+const DEFAULT_NOTIFICATION_FORMAT = 'PNG';
 const CONDITIONS = [
   {
     label: t('< (Smaller than)'),
@@ -326,33 +329,6 @@ const StyledNotificationAddButton = styled.div`
   }
 `;
 
-const StyledNotificationMethod = styled.div`
-  margin-bottom: 10px;
-
-  .input-container {
-    textarea {
-      height: auto;
-    }
-  }
-
-  .inline-container {
-    margin-bottom: 10px;
-
-    .input-container {
-      margin-left: 10px;
-    }
-
-    > div {
-      margin: 0;
-    }
-
-    .delete-button {
-      margin-left: 10px;
-      padding-top: 3px;
-    }
-  }
-`;
-
 type NotificationAddStatus = 'active' | 'disabled' | 'hidden';
 
 interface NotificationMethodAddProps {
@@ -392,116 +368,6 @@ type NotificationSetting = {
   options: NotificationMethod[];
 };
 
-interface NotificationMethodProps {
-  setting?: NotificationSetting | null;
-  index: number;
-  onUpdate?: (index: number, updatedSetting: NotificationSetting) => void;
-  onRemove?: (index: number) => void;
-}
-
-const NotificationMethod: FunctionComponent<NotificationMethodProps> = ({
-  setting = null,
-  index,
-  onUpdate,
-  onRemove,
-}) => {
-  const { method, recipients, options } = setting || {};
-  const [recipientValue, setRecipientValue] = useState<string>(
-    recipients || '',
-  );
-
-  if (!setting) {
-    return null;
-  }
-
-  const onMethodChange = (method: NotificationMethod) => {
-    // Since we're swapping the method, reset the recipients
-    setRecipientValue('');
-    if (onUpdate) {
-      const updatedSetting = {
-        ...setting,
-        method,
-        recipients: '',
-      };
-
-      onUpdate(index, updatedSetting);
-    }
-  };
-
-  const onRecipientsChange = (
-    event: React.ChangeEvent<HTMLTextAreaElement>,
-  ) => {
-    const { target } = event;
-
-    setRecipientValue(target.value);
-
-    if (onUpdate) {
-      const updatedSetting = {
-        ...setting,
-        recipients: target.value,
-      };
-
-      onUpdate(index, updatedSetting);
-    }
-  };
-
-  // Set recipients
-  if (!!recipients && recipientValue !== recipients) {
-    setRecipientValue(recipients);
-  }
-
-  const methodOptions = (options || []).map((method: NotificationMethod) => (
-    <Select.Option key={method} value={method}>
-      {t(method)}
-    </Select.Option>
-  ));
-
-  return (
-    <StyledNotificationMethod>
-      <div className="inline-container">
-        <StyledInputContainer>
-          <div className="input-container">
-            <Select
-              data-test="select-delivery-method"
-              onChange={onMethodChange}
-              placeholder="Select Delivery Method"
-              defaultValue={method}
-              value={method}
-            >
-              {methodOptions}
-            </Select>
-          </div>
-        </StyledInputContainer>
-        {method !== undefined && !!onRemove ? (
-          <span
-            role="button"
-            tabIndex={0}
-            className="delete-button"
-            onClick={() => onRemove(index)}
-          >
-            <Icon name="trash" />
-          </span>
-        ) : null}
-      </div>
-      {method !== undefined ? (
-        <StyledInputContainer>
-          <div className="control-label">{t(method)}</div>
-          <div className="input-container">
-            <textarea
-              name="recipients"
-              value={recipientValue}
-              onChange={onRecipientsChange}
-            />
-          </div>
-          <div className="helper">
-            {t('Recipients are separated by "," or ";"')}
-          </div>
-        </StyledInputContainer>
-      ) : null}
-    </StyledNotificationMethod>
-  );
-};
-
 const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
   addDangerToast,
   onAdd,
@@ -517,6 +383,9 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
   ] = useState<Partial<AlertObject> | null>();
   const [isHidden, setIsHidden] = useState<boolean>(true);
   const [contentType, setContentType] = useState<string>('dashboard');
+  const [reportFormat, setReportFormat] = useState<string>(
+    DEFAULT_NOTIFICATION_FORMAT,
+  );
 
   // Dropdown options
   const [conditionNotNull, setConditionNotNull] = useState<boolean>(false);
@@ -525,6 +394,9 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
   const [chartOptions, setChartOptions] = useState<MetaObject[]>([]);
 
   const isEditMode = alert !== null;
+  const formatOptionEnabled =
+    contentType === 'chart' &&
+    (isFeatureEnabled(FeatureFlag.ALERTS_ATTACH_REPORTS) || isReport);
 
   const [
     notificationAddState,
@@ -619,6 +491,10 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
         owner => (owner as MetaObject).value,
       ),
       recipients,
+      report_format:
+        contentType === 'dashboard'
+          ? DEFAULT_NOTIFICATION_FORMAT
+          : reportFormat || DEFAULT_NOTIFICATION_FORMAT,
     };
 
     if (data.recipients && !data.recipients.length) {
@@ -913,6 +789,12 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
     setContentType(target.value);
   };
 
+  const onFormatChange = (event: any) => {
+    const { target } = event;
+
+    setReportFormat(target.value);
+  };
+
   // Make sure notification settings has the required info
   const checkNotificationSettings = () => {
     if (!notificationSettings.length) {
@@ -985,22 +867,29 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
   useEffect(() => {
     if (resource) {
       // Add notification settings
-      const settings = (resource.recipients || []).map(setting => ({
-        method: setting.type as NotificationMethod,
-        // @ts-ignore: Type not assignable
-        recipients:
+      const settings = (resource.recipients || []).map(setting => {
+        const config =
           typeof setting.recipient_config_json === 'string'
-            ? (JSON.parse(setting.recipient_config_json) || {}).target
-            : setting.recipient_config_json,
-        options: NOTIFICATION_METHODS as NotificationMethod[], // Need better logic for this
-      }));
+            ? JSON.parse(setting.recipient_config_json)
+            : {};
+        return {
+          method: setting.type as NotificationMethod,
+          // @ts-ignore: Type not assignable
+          recipients: config.target || setting.recipient_config_json,
+          options: NOTIFICATION_METHODS as NotificationMethod[], // Need better logic for this
+        };
+      });
 
       setNotificationSettings(settings);
       setNotificationAddState(
         settings.length === NOTIFICATION_METHODS.length ? 'hidden' : 'active',
       );
       setContentType(resource.chart ? 'chart' : 'dashboard');
-
+      setReportFormat(
+        resource.chart
+          ? resource.report_format || DEFAULT_NOTIFICATION_FORMAT
+          : DEFAULT_NOTIFICATION_FORMAT,
+      );
       const validatorConfig =
         typeof resource.validator_config_json === 'string'
           ? JSON.parse(resource.validator_config_json)
@@ -1341,6 +1230,14 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
                 <Radio value="chart">Chart</Radio>
               </Radio.Group>
             </div>
+            {formatOptionEnabled && (
+              <div className="inline-container add-margin">
+                <Radio.Group onChange={onFormatChange} value={reportFormat}>
+                  <Radio value="PNG">PNG</Radio>
+                  <Radio value="CSV">CSV</Radio>
+                </Radio.Group>
+              </div>
+            )}
             <AsyncSelect
               className={
                 contentType === 'chart'
@@ -1385,18 +1282,14 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
               <h4>{t('Notification method')}</h4>
               <span className="required">*</span>
             </StyledSectionTitle>
-            <NotificationMethod
-              setting={notificationSettings[0]}
-              index={0}
-              onUpdate={updateNotificationSetting}
-              onRemove={removeNotificationSetting}
-            />
-            <NotificationMethod
-              setting={notificationSettings[1]}
-              index={1}
-              onUpdate={updateNotificationSetting}
-              onRemove={removeNotificationSetting}
-            />
+            {notificationSettings.map((notificationSetting, i) => (
+              <NotificationMethod
+                setting={notificationSetting}
+                index={i}
+                onUpdate={updateNotificationSetting}
+                onRemove={removeNotificationSetting}
+              />
+            ))}
             <NotificationMethodAdd
               data-test="notification-add"
               status={notificationAddState}
diff --git a/superset-frontend/src/views/CRUD/alert/components/NotificationMethod.tsx b/superset-frontend/src/views/CRUD/alert/components/NotificationMethod.tsx
new file mode 100644
index 000000000..7a43d2d59
--- /dev/null
+++ b/superset-frontend/src/views/CRUD/alert/components/NotificationMethod.tsx
@@ -0,0 +1,168 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import React, { FunctionComponent, useState } from 'react';
+import { styled, t } from '@superset-ui/core';
+import { NativeGraySelect as Select } from 'src/components/Select';
+import Icon from 'src/components/Icon';
+import { StyledInputContainer } from '../AlertReportModal';
+
+const StyledNotificationMethod = styled.div`
+  margin-bottom: 10px;
+
+  .input-container {
+    textarea {
+      height: auto;
+    }
+  }
+
+  .inline-container {
+    margin-bottom: 10px;
+
+    .input-container {
+      margin-left: 10px;
+    }
+
+    > div {
+      margin: 0;
+    }
+
+    .delete-button {
+      margin-left: 10px;
+      padding-top: 3px;
+    }
+  }
+`;
+
+type NotificationMethod = 'Email' | 'Slack';
+
+type NotificationSetting = {
+  method?: NotificationMethod;
+  recipients: string;
+  options: NotificationMethod[];
+};
+
+interface NotificationMethodProps {
+  setting?: NotificationSetting | null;
+  index: number;
+  onUpdate?: (index: number, updatedSetting: NotificationSetting) => void;
+  onRemove?: (index: number) => void;
+}
+
+export const NotificationMethod: FunctionComponent<NotificationMethodProps> = ({
+  setting = null,
+  index,
+  onUpdate,
+  onRemove,
+}) => {
+  const { method, recipients, options } = setting || {};
+  const [recipientValue, setRecipientValue] = useState<string>(
+    recipients || '',
+  );
+
+  if (!setting) {
+    return null;
+  }
+
+  const onMethodChange = (method: NotificationMethod) => {
+    // Since we're swapping the method, reset the recipients
+    setRecipientValue('');
+    if (onUpdate) {
+      const updatedSetting = {
+        ...setting,
+        method,
+        recipients: '',
+      };
+
+      onUpdate(index, updatedSetting);
+    }
+  };
+
+  const onRecipientsChange = (
+    event: React.ChangeEvent<HTMLTextAreaElement>,
+  ) => {
+    const { target } = event;
+
+    setRecipientValue(target.value);
+
+    if (onUpdate) {
+      const updatedSetting = {
+        ...setting,
+        recipients: target.value,
+      };
+
+      onUpdate(index, updatedSetting);
+    }
+  };
+
+  // Set recipients
+  if (!!recipients && recipientValue !== recipients) {
+    setRecipientValue(recipients);
+  }
+
+  const methodOptions = (options || []).map((method: NotificationMethod) => (
+    <Select.Option key={method} value={method}>
+      {t(method)}
+    </Select.Option>
+  ));
+
+  return (
+    <StyledNotificationMethod>
+      <div className="inline-container">
+        <StyledInputContainer>
+          <div className="input-container">
+            <Select
+              data-test="select-delivery-method"
+              onChange={onMethodChange}
+              placeholder="Select Delivery Method"
+              defaultValue={method}
+              value={method}
+            >
+              {methodOptions}
+            </Select>
+          </div>
+        </StyledInputContainer>
+        {method !== undefined && !!onRemove ? (
+          <span
+            role="button"
+            tabIndex={0}
+            className="delete-button"
+            onClick={() => onRemove(index)}
+          >
+            <Icon name="trash" />
+          </span>
+        ) : null}
+      </div>
+      {method !== undefined ? (
+        <StyledInputContainer>
+          <div className="control-label">{t(method)}</div>
+          <div className="input-container">
+            <textarea
+              name="recipients"
+              value={recipientValue}
+              onChange={onRecipientsChange}
+            />
+          </div>
+          <div className="helper">
+            {t('Recipients are separated by "," or ";"')}
+          </div>
+        </StyledInputContainer>
+      ) : null}
+    </StyledNotificationMethod>
+  );
+};
diff --git a/superset-frontend/src/views/CRUD/alert/types.ts b/superset-frontend/src/views/CRUD/alert/types.ts
index 556dbb986..6df150122 100644
--- a/superset-frontend/src/views/CRUD/alert/types.ts
+++ b/superset-frontend/src/views/CRUD/alert/types.ts
@@ -74,6 +74,7 @@ export type AlertObject = {
   owners?: Array<Owner | MetaObject>;
   sql?: string;
   recipients?: Array<Recipient>;
+  report_format?: 'PNG' | 'CSV';
   type?: string;
   validator_config_json?: {
     op?: Operator;
diff --git a/superset/migrations/versions/19e978e1b9c3_add_report_format_to_report_schedule_.py b/superset/migrations/versions/19e978e1b9c3_add_report_format_to_report_schedule_.py
new file mode 100644
index 000000000..ab25b88c5
--- /dev/null
+++ b/superset/migrations/versions/19e978e1b9c3_add_report_format_to_report_schedule_.py
@@ -0,0 +1,44 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""add_report_format_to_report_schedule_model.py
+
+Revision ID: 19e978e1b9c3
+Revises: fc3a3a8ff221
+Create Date: 2021-04-06 21:39:52.259223
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = "19e978e1b9c3"
+down_revision = "fc3a3a8ff221"
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+
+def upgrade():
+    op.add_column(
+        "report_schedule",
+        sa.Column(
+            "report_format", sa.String(length=50), server_default="PNG", nullable=True
+        ),
+    )
+
+
+def downgrade():
+    op.drop_column("report_schedule", "report_format")
diff --git a/superset/models/reports.py b/superset/models/reports.py
index be4073311..73c74ebc5 100644
--- a/superset/models/reports.py
+++ b/superset/models/reports.py
@@ -69,9 +69,9 @@ class ReportState(str, enum.Enum):
     GRACE = "On Grace"
 
 
-class ReportEmailFormat(str, enum.Enum):
-    VISUALIZATION = "Visualization"
-    DATA = "Raw data"
+class ReportDataFormat(str, enum.Enum):
+    VISUALIZATION = "PNG"
+    DATA = "CSV"
 
 
 report_schedule_user = Table(
@@ -102,6 +102,7 @@ class ReportSchedule(Model, AuditMixinNullable):
     context_markdown = Column(Text)
     active = Column(Boolean, default=True, index=True)
     crontab = Column(String(1000), nullable=False)
+    report_format = Column(String(50), default=ReportDataFormat.VISUALIZATION)
     sql = Column(Text())
     # (Alerts/Reports) M-O to chart
     chart_id = Column(Integer, ForeignKey("slices.id"), nullable=True)
diff --git a/superset/reports/api.py b/superset/reports/api.py
index d7205b42a..5fc2808b6 100644
--- a/superset/reports/api.py
+++ b/superset/reports/api.py
@@ -95,6 +95,7 @@ class ReportScheduleRestApi(BaseSupersetModelRestApi):
         "recipients.id",
         "recipients.recipient_config_json",
         "recipients.type",
+        "report_format",
         "sql",
         "type",
         "validator_config_json",
@@ -140,11 +141,12 @@ class ReportScheduleRestApi(BaseSupersetModelRestApi):
         "name",
         "owners",
         "recipients",
+        "report_format",
         "sql",
         "type",
-        "working_timeout",
         "validator_config_json",
         "validator_type",
+        "working_timeout",
     ]
     edit_columns = add_columns
     add_model_schema = ReportSchedulePostSchema()
diff --git a/superset/reports/commands/exceptions.py b/superset/reports/commands/exceptions.py
index b71068b46..dfe2402da 100644
--- a/superset/reports/commands/exceptions.py
+++ b/superset/reports/commands/exceptions.py
@@ -112,6 +112,10 @@ class ReportScheduleScreenshotFailedError(CommandException):
     message = _("Report Schedule execution failed when generating a screenshot.")
 
 
+class ReportScheduleCsvFailedError(CommandException):
+    message = _("Report Schedule execution failed when generating a csv.")
+
+
 class ReportScheduleExecuteUnexpectedError(CommandException):
     message = _("Report Schedule execution got an unexpected error.")
 
@@ -163,6 +167,10 @@ class ReportScheduleScreenshotTimeout(CommandException):
     message = _("A timeout occurred while taking a screenshot.")
 
 
+class ReportScheduleCsvTimeout(CommandException):
+    message = _("A timeout occurred while generating a csv.")
+
+
 class ReportScheduleAlertGracePeriodError(CommandException):
     message = _("Alert fired during grace period.")
 
diff --git a/superset/reports/commands/execute.py b/superset/reports/commands/execute.py
index ed9691cea..befd0b233 100644
--- a/superset/reports/commands/execute.py
+++ b/superset/reports/commands/execute.py
@@ -27,8 +27,9 @@ from sqlalchemy.orm import Session
 from superset import app
 from superset.commands.base import BaseCommand
 from superset.commands.exceptions import CommandException
-from superset.extensions import feature_flag_manager
+from superset.extensions import feature_flag_manager, machine_auth_provider_factory
 from superset.models.reports import (
+    ReportDataFormat,
     ReportExecutionLog,
     ReportRecipients,
     ReportRecipientType,
@@ -40,6 +41,8 @@ from superset.reports.commands.alert import AlertCommand
 from superset.reports.commands.exceptions import (
     ReportScheduleAlertEndGracePeriodError,
     ReportScheduleAlertGracePeriodError,
+    ReportScheduleCsvFailedError,
+    ReportScheduleCsvTimeout,
     ReportScheduleExecuteUnexpectedError,
     ReportScheduleNotFoundError,
     ReportScheduleNotificationError,
@@ -59,6 +62,7 @@ from superset.reports.notifications import create_notification
 from superset.reports.notifications.base import NotificationContent
 from superset.reports.notifications.exceptions import NotificationError
 from superset.utils.celery import session_scope
+from superset.utils.csv import get_chart_csv_data
 from superset.utils.screenshots import (
     BaseScreenshot,
     ChartScreenshot,
@@ -129,11 +133,19 @@ class BaseReportState:
         self._session.add(log)
         self._session.commit()
 
-    def _get_url(self, user_friendly: bool = False, **kwargs: Any) -> str:
+    def _get_url(
+        self, user_friendly: bool = False, csv: bool = False, **kwargs: Any
+    ) -> str:
         """
         Get the url for this report schedule: chart or dashboard
         """
         if self._report_schedule.chart:
+            if csv:
+                return get_url_path(
+                    "Superset.explore_json",
+                    csv="true",
+                    form_data=json.dumps({"slice_id": self._report_schedule.chart_id}),
+                )
             return get_url_path(
                 "Superset.slice",
                 user_friendly=user_friendly,
@@ -147,7 +159,7 @@ class BaseReportState:
             **kwargs,
         )
 
-    def _get_screenshot_user(self) -> User:
+    def _get_user(self) -> User:
         user = (
             self._session.query(User)
             .filter(User.username == app.config["THUMBNAIL_SELENIUM_USER"])
@@ -180,7 +192,7 @@ class BaseReportState:
                 window_size=app.config["WEBDRIVER_WINDOW"]["dashboard"],
                 thumb_size=app.config["WEBDRIVER_WINDOW"]["dashboard"],
             )
-        user = self._get_screenshot_user()
+        user = self._get_user()
         try:
             image_data = screenshot.get_screenshot(user=user)
         except SoftTimeLimitExceeded:
@@ -194,23 +206,50 @@ class BaseReportState:
             raise ReportScheduleScreenshotFailedError()
         return image_data
 
+    def _get_csv_data(self) -> bytes:
+        if self._report_schedule.chart:
+            url = self._get_url(csv=True)
+            auth_cookies = machine_auth_provider_factory.instance.get_auth_cookies(
+                self._get_user()
+            )
+        try:
+            csv_data = get_chart_csv_data(url, auth_cookies)
+        except SoftTimeLimitExceeded:
+            raise ReportScheduleCsvTimeout()
+        except Exception as ex:
+            raise ReportScheduleCsvFailedError(f"Failed generating csv {str(ex)}")
+        if not csv_data:
+            raise ReportScheduleCsvFailedError()
+        return csv_data
+
     def _get_notification_content(self) -> NotificationContent:
         """
         Gets a notification content, this is composed by a title and a screenshot
 
         :raises: ReportScheduleScreenshotFailedError
         """
+        csv_data = None
+        error_text = None
         screenshot_data = None
         url = self._get_url(user_friendly=True)
         if (
             feature_flag_manager.is_feature_enabled("ALERTS_ATTACH_REPORTS")
             or self._report_schedule.type == ReportScheduleType.REPORT
         ):
-            screenshot_data = self._get_screenshot()
-            if not screenshot_data:
+            if self._report_schedule.report_format == ReportDataFormat.VISUALIZATION:
+                screenshot_data = self._get_screenshot()
+                if not screenshot_data:
+                    error_text = "Unexpected missing screenshot"
+            elif (
+                self._report_schedule.chart
+                and self._report_schedule.report_format == ReportDataFormat.DATA
+            ):
+                csv_data = self._get_csv_data()
+                if not csv_data:
+                    error_text = "Unexpected missing csv file"
+            if error_text:
                 return NotificationContent(
-                    name=self._report_schedule.name,
-                    text="Unexpected missing screenshot",
+                    name=self._report_schedule.name, text=error_text
                 )
 
         if self._report_schedule.chart:
@@ -228,6 +267,7 @@ class BaseReportState:
             url=url,
             screenshot=screenshot_data,
             description=self._report_schedule.description,
+            csv=csv_data,
         )
 
     @staticmethod
diff --git a/superset/reports/notifications/base.py b/superset/reports/notifications/base.py
index 60f5f0a0e..d8b377261 100644
--- a/superset/reports/notifications/base.py
+++ b/superset/reports/notifications/base.py
@@ -24,10 +24,11 @@ from superset.models.reports import ReportRecipients, ReportRecipientType
 @dataclass
 class NotificationContent:
     name: str
-    url: Optional[str] = None  # url to chart/dashboard for this screenshot
+    csv: Optional[bytes] = None  # bytes for csv file
     screenshot: Optional[bytes] = None  # bytes for the screenshot
     text: Optional[str] = None
     description: Optional[str] = ""
+    url: Optional[str] = None  # url to chart/dashboard for this screenshot
 
 
 class BaseNotification:  # pylint: disable=too-few-public-methods
diff --git a/superset/reports/notifications/email.py b/superset/reports/notifications/email.py
index 09023f92d..3f2296c3b 100644
--- a/superset/reports/notifications/email.py
+++ b/superset/reports/notifications/email.py
@@ -19,7 +19,7 @@ import json
 import logging
 from dataclasses import dataclass
 from email.utils import make_msgid, parseaddr
-from typing import Dict, Optional
+from typing import Any, Dict, Optional
 
 from flask_babel import gettext as __
 
@@ -35,6 +35,7 @@ logger = logging.getLogger(__name__)
 @dataclass
 class EmailContent:
     body: str
+    data: Optional[Dict[str, Any]] = None
     images: Optional[Dict[str, bytes]] = None
 
 
@@ -64,22 +65,24 @@ class EmailNotification(BaseNotification):  # pylint: disable=too-few-public-met
         # Get the domain from the 'From' address ..
         # and make a message id without the < > in the end
         image = None
+        csv_data = None
         domain = self._get_smtp_domain()
         msgid = make_msgid(domain)[1:-1]
         body = __(
             """
             <p>%(description)s</p>
             <b><a href="%(url)s">Explore in Superset</a></b><p></p>
-            <img src="cid:%(msgid)s">
+            %(img_tag)s
             """,
             description=self._content.description or "",
             url=self._content.url,
-            msgid=msgid,
+            img_tag=f'<img src="cid:{msgid}">' if self._content.screenshot else "",
         )
         if self._content.screenshot:
             image = {msgid: self._content.screenshot}
-
-        return EmailContent(body=body, images=image)
+        if self._content.csv:
+            csv_data = {__("%(name)s.csv", name=self._content.name): self._content.csv}
+        return EmailContent(body=body, images=image, data=csv_data)
 
     def _get_subject(self) -> str:
         return __(
@@ -102,7 +105,7 @@ class EmailNotification(BaseNotification):  # pylint: disable=too-few-public-met
                 content.body,
                 app.config,
                 files=[],
-                data=None,
+                data=content.data,
                 images=content.images,
                 bcc="",
                 mime_subtype="related",
diff --git a/superset/reports/notifications/slack.py b/superset/reports/notifications/slack.py
index 97333dd77..4fd9fb35f 100644
--- a/superset/reports/notifications/slack.py
+++ b/superset/reports/notifications/slack.py
@@ -72,16 +72,20 @@ class SlackNotification(BaseNotification):  # pylint: disable=too-few-public-met
             url=self._content.url,
         )
 
-    def _get_inline_screenshot(self) -> Optional[Union[str, IOBase, bytes]]:
+    def _get_inline_file(self) -> Optional[Union[str, IOBase, bytes]]:
+        if self._content.csv:
+            return self._content.csv
         if self._content.screenshot:
             return self._content.screenshot
         return None
 
     @retry(SlackApiError, delay=10, backoff=2, tries=5)
     def send(self) -> None:
-        file = self._get_inline_screenshot()
+        file = self._get_inline_file()
+        title = self._content.name
         channel = self._get_channel()
         body = self._get_body()
+        file_type = "csv" if self._content.csv else "png"
         try:
             token = app.config["SLACK_API_TOKEN"]
             if callable(token):
@@ -90,7 +94,11 @@ class SlackNotification(BaseNotification):  # pylint: disable=too-few-public-met
             # files_upload returns SlackResponse as we run it in sync mode.
             if file:
                 client.files_upload(
-                    channels=channel, file=file, initial_comment=body, title="subject",
+                    channels=channel,
+                    file=file,
+                    initial_comment=body,
+                    title=title,
+                    filetype=file_type,
                 )
             else:
                 client.chat_postMessage(channel=channel, text=body)
diff --git a/superset/reports/schemas.py b/superset/reports/schemas.py
index 58eab2ce5..1ea711f4c 100644
--- a/superset/reports/schemas.py
+++ b/superset/reports/schemas.py
@@ -22,6 +22,7 @@ from marshmallow import fields, Schema, validate, validates_schema
 from marshmallow.validate import Length, Range, ValidationError
 
 from superset.models.reports import (
+    ReportDataFormat,
     ReportRecipientType,
     ReportScheduleType,
     ReportScheduleValidatorType,
@@ -178,6 +179,10 @@ class ReportSchedulePostSchema(Schema):
     )
 
     recipients = fields.List(fields.Nested(ReportRecipientSchema))
+    report_format = fields.String(
+        default=ReportDataFormat.VISUALIZATION,
+        validate=validate.OneOf(choices=tuple(key.value for key in ReportDataFormat)),
+    )
 
     @validates_schema
     def validate_report_references(  # pylint: disable=unused-argument,no-self-use
@@ -253,3 +258,7 @@ class ReportSchedulePutSchema(Schema):
         validate=[Range(min=1, error=_("Value must be greater than 0"))],
     )
     recipients = fields.List(fields.Nested(ReportRecipientSchema), required=False)
+    report_format = fields.String(
+        default=ReportDataFormat.VISUALIZATION,
+        validate=validate.OneOf(choices=tuple(key.value for key in ReportDataFormat)),
+    )
diff --git a/superset/utils/csv.py b/superset/utils/csv.py
index d91d427d5..3e5fe4621 100644
--- a/superset/utils/csv.py
+++ b/superset/utils/csv.py
@@ -15,7 +15,9 @@
 # specific language governing permissions and limitations
 # under the License.
 import re
-from typing import Any
+import urllib.request
+from typing import Any, Dict, Optional
+from urllib.error import URLError
 
 import pandas as pd
 
@@ -65,3 +67,20 @@ def df_to_escaped_csv(df: pd.DataFrame, **kwargs: Any) -> Any:
     df = df.applymap(escape_values)
 
     return df.to_csv(**kwargs)
+
+
+def get_chart_csv_data(
+    chart_url: str, auth_cookies: Optional[Dict[str, str]] = None
+) -> Optional[bytes]:
+    content = None
+    if auth_cookies:
+        opener = urllib.request.build_opener()
+        cookie_str = ";".join([f"{key}={val}" for key, val in auth_cookies.items()])
+        opener.addheaders.append(("Cookie", cookie_str))
+        response = opener.open(chart_url)
+        content = response.read()
+    if response.getcode() != 200:
+        raise URLError(response.getcode())
+    if content:
+        return content
+    return None
diff --git a/tests/reports/commands_tests.py b/tests/reports/commands_tests.py
index 7a20ab3c8..8392d7417 100644
--- a/tests/reports/commands_tests.py
+++ b/tests/reports/commands_tests.py
@@ -30,6 +30,7 @@ from superset import db, security_manager
 from superset.models.core import Database
 from superset.models.dashboard import Dashboard
 from superset.models.reports import (
+    ReportDataFormat,
     ReportExecutionLog,
     ReportRecipients,
     ReportRecipientType,
@@ -44,6 +45,8 @@ from superset.reports.commands.exceptions import (
     AlertQueryInvalidTypeError,
     AlertQueryMultipleColumnsError,
     AlertQueryMultipleRowsError,
+    ReportScheduleCsvFailedError,
+    ReportScheduleCsvTimeout,
     ReportScheduleNotFoundError,
     ReportScheduleNotificationError,
     ReportSchedulePreviousWorkingError,
@@ -67,7 +70,10 @@ pytestmark = pytest.mark.usefixtures(
     "load_world_bank_dashboard_with_slices_module_scope"
 )
 
-test_id = str(uuid4())
+TEST_ID = str(uuid4())
+CSV_FILE = read_fixture("trends.csv")
+SCREENSHOT_FILE = read_fixture("sample.png")
+OWNER_EMAIL = "admin@fab.org"
 
 
 def get_target_from_report_schedule(report_schedule: ReportSchedule) -> List[str]:
@@ -128,13 +134,14 @@ def create_report_notification(
     validator_type: Optional[str] = None,
     validator_config_json: Optional[str] = None,
     grace_period: Optional[int] = None,
+    report_format: Optional[ReportDataFormat] = None,
 ) -> ReportSchedule:
     report_type = report_type or ReportScheduleType.REPORT
     target = email_target or slack_channel
     config_json = {"target": target}
     owner = (
         db.session.query(security_manager.user_model)
-        .filter_by(email="admin@fab.org")
+        .filter_by(email=OWNER_EMAIL)
         .one_or_none()
     )
 
@@ -151,7 +158,7 @@ def create_report_notification(
 
     report_schedule = insert_report_schedule(
         type=report_type,
-        name=f"report",
+        name=f"report_with_csv" if report_format else f"report",
         crontab=f"0 9 * * *",
         description=f"Daily report",
         sql=sql,
@@ -163,6 +170,7 @@ def create_report_notification(
         validator_type=validator_type,
         validator_config_json=validator_config_json,
         grace_period=grace_period,
+        report_format=report_format or ReportDataFormat.VISUALIZATION,
     )
     return report_schedule
 
@@ -207,6 +215,19 @@ def create_report_email_chart():
         cleanup_report_schedule(report_schedule)
 
 
+@pytest.fixture()
+def create_report_email_chart_with_csv():
+    with app.app_context():
+        chart = db.session.query(Slice).first()
+        report_schedule = create_report_notification(
+            email_target="target@email.com",
+            chart=chart,
+            report_format=ReportDataFormat.DATA,
+        )
+        yield report_schedule
+        cleanup_report_schedule(report_schedule)
+
+
 @pytest.fixture()
 def create_report_email_dashboard():
     with app.app_context():
@@ -231,6 +252,20 @@ def create_report_slack_chart():
         cleanup_report_schedule(report_schedule)
 
 
+@pytest.fixture()
+def create_report_slack_chart_with_csv():
+    with app.app_context():
+        chart = db.session.query(Slice).first()
+        report_schedule = create_report_notification(
+            slack_channel="slack_channel",
+            chart=chart,
+            report_format=ReportDataFormat.DATA,
+        )
+        yield report_schedule
+
+        cleanup_report_schedule(report_schedule)
+
+
 @pytest.fixture()
 def create_report_slack_chart_working():
     with app.app_context():
@@ -309,7 +344,7 @@ def create_alert_slack_chart_grace():
         cleanup_report_schedule(report_schedule)
 
 
-@pytest.yield_fixture(
+@pytest.fixture(
     params=["alert1", "alert2", "alert3", "alert4", "alert5", "alert6", "alert7",]
 )
 def create_alert_email_chart(request):
@@ -371,7 +406,7 @@ def create_alert_email_chart(request):
             cleanup_report_schedule(report_schedule)
 
 
-@pytest.yield_fixture(
+@pytest.fixture(
     params=[
         "alert1",
         "alert2",
@@ -453,7 +488,7 @@ def create_no_alert_email_chart(request):
             cleanup_report_schedule(report_schedule)
 
 
-@pytest.yield_fixture(params=["alert1", "alert2"])
+@pytest.fixture(params=["alert1", "alert2"])
 def create_mul_alert_email_chart(request):
     param_config = {
         "alert1": {
@@ -488,7 +523,7 @@ def create_mul_alert_email_chart(request):
             cleanup_report_schedule(report_schedule)
 
 
-@pytest.yield_fixture(params=["alert1", "alert2"])
+@pytest.fixture(params=["alert1", "alert2"])
 def create_invalid_sql_alert_email_chart(request):
     param_config = {
         "alert1": {
@@ -530,18 +565,17 @@ def create_invalid_sql_alert_email_chart(request):
 @patch("superset.reports.notifications.email.send_email_smtp")
 @patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
 def test_email_chart_report_schedule(
-    screenshot_mock, email_mock, create_report_email_chart
+    screenshot_mock, email_mock, create_report_email_chart,
 ):
     """
-    ExecuteReport Command: Test chart email report schedule
+    ExecuteReport Command: Test chart email report schedule with
     """
     # setup screenshot mock
-    screenshot = read_fixture("sample.png")
-    screenshot_mock.return_value = screenshot
+    screenshot_mock.return_value = SCREENSHOT_FILE
 
     with freeze_time("2020-01-01T00:00:00Z"):
         AsyncExecuteReportScheduleCommand(
-            test_id, create_report_email_chart.id, datetime.utcnow()
+            TEST_ID, create_report_email_chart.id, datetime.utcnow()
         ).run()
 
         notification_targets = get_target_from_report_schedule(
@@ -557,7 +591,50 @@ def test_email_chart_report_schedule(
         assert email_mock.call_args[0][0] == notification_targets[0]
         # Assert the email inline screenshot
         smtp_images = email_mock.call_args[1]["images"]
-        assert smtp_images[list(smtp_images.keys())[0]] == screenshot
+        assert smtp_images[list(smtp_images.keys())[0]] == SCREENSHOT_FILE
+        # Assert logs are correct
+        assert_log(ReportState.SUCCESS)
+
+
+@pytest.mark.usefixtures(
+    "load_birth_names_dashboard_with_slices", "create_report_email_chart_with_csv"
+)
+@patch("superset.utils.csv.urllib.request.urlopen")
+@patch("superset.utils.csv.urllib.request.OpenerDirector.open")
+@patch("superset.reports.notifications.email.send_email_smtp")
+@patch("superset.utils.csv.get_chart_csv_data")
+def test_email_chart_report_schedule_with_csv(
+    csv_mock, email_mock, mock_open, mock_urlopen, create_report_email_chart_with_csv,
+):
+    """
+    ExecuteReport Command: Test chart email report schedule with CSV
+    """
+    # setup csv mock
+    response = Mock()
+    mock_open.return_value = response
+    mock_urlopen.return_value = response
+    mock_urlopen.return_value.getcode.return_value = 200
+    response.read.return_value = CSV_FILE
+
+    with freeze_time("2020-01-01T00:00:00Z"):
+        AsyncExecuteReportScheduleCommand(
+            TEST_ID, create_report_email_chart_with_csv.id, datetime.utcnow()
+        ).run()
+
+        notification_targets = get_target_from_report_schedule(
+            create_report_email_chart_with_csv
+        )
+        # assert that the link sent is correct
+        assert (
+            f'<a href="http://0.0.0.0:8080/superset/slice/'
+            f'{create_report_email_chart_with_csv.chart.id}/">Explore in Superset</a>'
+            in email_mock.call_args[0][2]
+        )
+        # Assert the email smtp address
+        assert email_mock.call_args[0][0] == notification_targets[0]
+        # Assert the email csv file
+        smtp_images = email_mock.call_args[1]["data"]
+        assert smtp_images[list(smtp_images.keys())[0]] == CSV_FILE
         # Assert logs are correct
         assert_log(ReportState.SUCCESS)
 
@@ -574,12 +651,11 @@ def test_email_dashboard_report_schedule(
     ExecuteReport Command: Test dashboard email report schedule
     """
     # setup screenshot mock
-    screenshot = read_fixture("sample.png")
-    screenshot_mock.return_value = screenshot
+    screenshot_mock.return_value = SCREENSHOT_FILE
 
     with freeze_time("2020-01-01T00:00:00Z"):
         AsyncExecuteReportScheduleCommand(
-            test_id, create_report_email_dashboard.id, datetime.utcnow()
+            TEST_ID, create_report_email_dashboard.id, datetime.utcnow()
         ).run()
 
         notification_targets = get_target_from_report_schedule(
@@ -589,7 +665,7 @@ def test_email_dashboard_report_schedule(
         assert email_mock.call_args[0][0] == notification_targets[0]
         # Assert the email inline screenshot
         smtp_images = email_mock.call_args[1]["images"]
-        assert smtp_images[list(smtp_images.keys())[0]] == screenshot
+        assert smtp_images[list(smtp_images.keys())[0]] == SCREENSHOT_FILE
         # Assert logs are correct
         assert_log(ReportState.SUCCESS)
 
@@ -600,25 +676,63 @@ def test_email_dashboard_report_schedule(
 @patch("superset.reports.notifications.slack.WebClient.files_upload")
 @patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
 def test_slack_chart_report_schedule(
-    screenshot_mock, file_upload_mock, create_report_slack_chart
+    screenshot_mock, file_upload_mock, create_report_slack_chart,
 ):
     """
     ExecuteReport Command: Test chart slack report schedule
     """
     # setup screenshot mock
-    screenshot = read_fixture("sample.png")
-    screenshot_mock.return_value = screenshot
+    screenshot_mock.return_value = SCREENSHOT_FILE
 
     with freeze_time("2020-01-01T00:00:00Z"):
         AsyncExecuteReportScheduleCommand(
-            test_id, create_report_slack_chart.id, datetime.utcnow()
+            TEST_ID, create_report_slack_chart.id, datetime.utcnow()
         ).run()
 
         notification_targets = get_target_from_report_schedule(
             create_report_slack_chart
         )
         assert file_upload_mock.call_args[1]["channels"] == notification_targets[0]
-        assert file_upload_mock.call_args[1]["file"] == screenshot
+        assert file_upload_mock.call_args[1]["file"] == SCREENSHOT_FILE
+
+        # Assert logs are correct
+        assert_log(ReportState.SUCCESS)
+
+
+@pytest.mark.usefixtures(
+    "load_birth_names_dashboard_with_slices", "create_report_slack_chart_with_csv"
+)
+@patch("superset.reports.notifications.slack.WebClient.files_upload")
+@patch("superset.utils.csv.urllib.request.urlopen")
+@patch("superset.utils.csv.urllib.request.OpenerDirector.open")
+@patch("superset.utils.csv.get_chart_csv_data")
+def test_slack_chart_report_schedule_with_csv(
+    csv_mock,
+    mock_open,
+    mock_urlopen,
+    file_upload_mock,
+    create_report_slack_chart_with_csv,
+):
+    """
+    ExecuteReport Command: Test chart slack report schedule with CSV
+    """
+    # setup csv mock
+    response = Mock()
+    mock_open.return_value = response
+    mock_urlopen.return_value = response
+    mock_urlopen.return_value.getcode.return_value = 200
+    response.read.return_value = CSV_FILE
+
+    with freeze_time("2020-01-01T00:00:00Z"):
+        AsyncExecuteReportScheduleCommand(
+            TEST_ID, create_report_slack_chart_with_csv.id, datetime.utcnow()
+        ).run()
+
+        notification_targets = get_target_from_report_schedule(
+            create_report_slack_chart_with_csv
+        )
+        assert file_upload_mock.call_args[1]["channels"] == notification_targets[0]
+        assert file_upload_mock.call_args[1]["file"] == CSV_FILE
 
         # Assert logs are correct
         assert_log(ReportState.SUCCESS)
@@ -631,7 +745,7 @@ def test_report_schedule_not_found(create_report_slack_chart):
     """
     max_id = db.session.query(func.max(ReportSchedule.id)).scalar()
     with pytest.raises(ReportScheduleNotFoundError):
-        AsyncExecuteReportScheduleCommand(test_id, max_id + 1, datetime.utcnow()).run()
+        AsyncExecuteReportScheduleCommand(TEST_ID, max_id + 1, datetime.utcnow()).run()
 
 
 @pytest.mark.usefixtures("create_report_slack_chart_working")
@@ -643,7 +757,7 @@ def test_report_schedule_working(create_report_slack_chart_working):
     with freeze_time("2020-01-01T00:00:00Z"):
         with pytest.raises(ReportSchedulePreviousWorkingError):
             AsyncExecuteReportScheduleCommand(
-                test_id, create_report_slack_chart_working.id, datetime.utcnow()
+                TEST_ID, create_report_slack_chart_working.id, datetime.utcnow()
             ).run()
 
         assert_log(
@@ -665,7 +779,7 @@ def test_report_schedule_working_timeout(create_report_slack_chart_working):
 
         with pytest.raises(ReportScheduleWorkingTimeoutError):
             AsyncExecuteReportScheduleCommand(
-                test_id, create_report_slack_chart_working.id, datetime.utcnow()
+                TEST_ID, create_report_slack_chart_working.id, datetime.utcnow()
             ).run()
 
     # Only needed for MySQL, understand why
@@ -691,7 +805,7 @@ def test_report_schedule_success_grace(create_alert_slack_chart_success):
 
     with freeze_time(current_time):
         AsyncExecuteReportScheduleCommand(
-            test_id, create_alert_slack_chart_success.id, datetime.utcnow()
+            TEST_ID, create_alert_slack_chart_success.id, datetime.utcnow()
         ).run()
 
     db.session.commit()
@@ -710,7 +824,7 @@ def test_report_schedule_success_grace_end(create_alert_slack_chart_grace):
 
     with freeze_time(current_time):
         AsyncExecuteReportScheduleCommand(
-            test_id, create_alert_slack_chart_grace.id, datetime.utcnow()
+            TEST_ID, create_alert_slack_chart_grace.id, datetime.utcnow()
         ).run()
 
     db.session.commit()
@@ -720,10 +834,13 @@ def test_report_schedule_success_grace_end(create_alert_slack_chart_grace):
 @pytest.mark.usefixtures("create_alert_email_chart")
 @patch("superset.reports.notifications.email.send_email_smtp")
 @patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
-def test_alert_limit_is_applied(screenshot_mock, email_mock, create_alert_email_chart):
+def test_alert_limit_is_applied(
+    screenshot_mock, email_mock, create_alert_email_chart,
+):
     """
     ExecuteReport Command: Test that all alerts apply a SQL limit to stmts
     """
+    screenshot_mock.return_value = SCREENSHOT_FILE
 
     with patch.object(
         create_alert_email_chart.database.db_engine_spec, "execute", return_value=None
@@ -734,7 +851,7 @@ def test_alert_limit_is_applied(screenshot_mock, email_mock, create_alert_email_
             return_value=None,
         ) as fetch_data_mock:
             AsyncExecuteReportScheduleCommand(
-                test_id, create_alert_email_chart.id, datetime.utcnow()
+                TEST_ID, create_alert_email_chart.id, datetime.utcnow()
             ).run()
             assert "LIMIT 2" in execute_mock.call_args[0][1]
 
@@ -753,13 +870,12 @@ def test_email_dashboard_report_fails(
     # setup screenshot mock
     from smtplib import SMTPException
 
-    screenshot = read_fixture("sample.png")
-    screenshot_mock.return_value = screenshot
+    screenshot_mock.return_value = SCREENSHOT_FILE
     email_mock.side_effect = SMTPException("Could not connect to SMTP XPTO")
 
     with pytest.raises(ReportScheduleNotificationError):
         AsyncExecuteReportScheduleCommand(
-            test_id, create_report_email_dashboard.id, datetime.utcnow()
+            TEST_ID, create_report_email_dashboard.id, datetime.utcnow()
         ).run()
 
     assert_log(ReportState.ERROR, error_message="Could not connect to SMTP XPTO")
@@ -774,17 +890,18 @@ def test_email_dashboard_report_fails(
     "superset.extensions.feature_flag_manager._feature_flags",
     ALERTS_ATTACH_REPORTS=True,
 )
-def test_slack_chart_alert(screenshot_mock, email_mock, create_alert_email_chart):
+def test_slack_chart_alert(
+    screenshot_mock, email_mock, create_alert_email_chart,
+):
     """
     ExecuteReport Command: Test chart slack alert
     """
     # setup screenshot mock
-    screenshot = read_fixture("sample.png")
-    screenshot_mock.return_value = screenshot
+    screenshot_mock.return_value = SCREENSHOT_FILE
 
     with freeze_time("2020-01-01T00:00:00Z"):
         AsyncExecuteReportScheduleCommand(
-            test_id, create_alert_email_chart.id, datetime.utcnow()
+            TEST_ID, create_alert_email_chart.id, datetime.utcnow()
         ).run()
 
         notification_targets = get_target_from_report_schedule(create_alert_email_chart)
@@ -792,7 +909,7 @@ def test_slack_chart_alert(screenshot_mock, email_mock, create_alert_email_chart
         assert email_mock.call_args[0][0] == notification_targets[0]
         # Assert the email inline screenshot
         smtp_images = email_mock.call_args[1]["images"]
-        assert smtp_images[list(smtp_images.keys())[0]] == screenshot
+        assert smtp_images[list(smtp_images.keys())[0]] == SCREENSHOT_FILE
         # Assert logs are correct
         assert_log(ReportState.SUCCESS)
 
@@ -813,7 +930,7 @@ def test_slack_chart_alert_no_attachment(email_mock, create_alert_email_chart):
 
     with freeze_time("2020-01-01T00:00:00Z"):
         AsyncExecuteReportScheduleCommand(
-            test_id, create_alert_email_chart.id, datetime.utcnow()
+            TEST_ID, create_alert_email_chart.id, datetime.utcnow()
         ).run()
 
         notification_targets = get_target_from_report_schedule(create_alert_email_chart)
@@ -831,7 +948,7 @@ def test_slack_chart_alert_no_attachment(email_mock, create_alert_email_chart):
 @patch("superset.reports.notifications.slack.WebClient")
 @patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
 def test_slack_token_callable_chart_report(
-    screenshot_mock, slack_client_mock_class, create_report_slack_chart
+    screenshot_mock, slack_client_mock_class, create_report_slack_chart,
 ):
     """
     ExecuteReport Command: Test chart slack alert (slack token callable)
@@ -839,12 +956,11 @@ def test_slack_token_callable_chart_report(
     slack_client_mock_class.return_value = Mock()
     app.config["SLACK_API_TOKEN"] = Mock(return_value="cool_code")
     # setup screenshot mock
-    screenshot = read_fixture("sample.png")
-    screenshot_mock.return_value = screenshot
+    screenshot_mock.return_value = SCREENSHOT_FILE
 
     with freeze_time("2020-01-01T00:00:00Z"):
         AsyncExecuteReportScheduleCommand(
-            test_id, create_report_slack_chart.id, datetime.utcnow()
+            TEST_ID, create_report_slack_chart.id, datetime.utcnow()
         ).run()
         app.config["SLACK_API_TOKEN"].assert_called_once()
         assert slack_client_mock_class.called_with(token="cool_code", proxy="")
@@ -858,7 +974,7 @@ def test_email_chart_no_alert(create_no_alert_email_chart):
     """
     with freeze_time("2020-01-01T00:00:00Z"):
         AsyncExecuteReportScheduleCommand(
-            test_id, create_no_alert_email_chart.id, datetime.utcnow()
+            TEST_ID, create_no_alert_email_chart.id, datetime.utcnow()
         ).run()
     assert_log(ReportState.NOOP)
 
@@ -873,7 +989,7 @@ def test_email_mul_alert(create_mul_alert_email_chart):
             (AlertQueryMultipleRowsError, AlertQueryMultipleColumnsError)
         ):
             AsyncExecuteReportScheduleCommand(
-                test_id, create_mul_alert_email_chart.id, datetime.utcnow()
+                TEST_ID, create_mul_alert_email_chart.id, datetime.utcnow()
             ).run()
 
 
@@ -886,6 +1002,7 @@ def test_soft_timeout_alert(email_mock, create_alert_email_chart):
     ExecuteReport Command: Test soft timeout on alert queries
     """
     from celery.exceptions import SoftTimeLimitExceeded
+
     from superset.reports.commands.exceptions import AlertQueryTimeout
 
     with patch.object(
@@ -894,12 +1011,12 @@ def test_soft_timeout_alert(email_mock, create_alert_email_chart):
         execute_mock.side_effect = SoftTimeLimitExceeded()
         with pytest.raises(AlertQueryTimeout):
             AsyncExecuteReportScheduleCommand(
-                test_id, create_alert_email_chart.id, datetime.utcnow()
+                TEST_ID, create_alert_email_chart.id, datetime.utcnow()
             ).run()
 
     notification_targets = get_target_from_report_schedule(create_alert_email_chart)
     # Assert the email smtp address, asserts a notification was sent with the error
-    assert email_mock.call_args[0][0] == "admin@fab.org"
+    assert email_mock.call_args[0][0] == OWNER_EMAIL
 
     assert_log(
         ReportState.ERROR, error_message="A timeout occurred while executing the query."
@@ -920,22 +1037,93 @@ def test_soft_timeout_screenshot(screenshot_mock, email_mock, create_alert_email
     ExecuteReport Command: Test soft timeout on screenshot
     """
     from celery.exceptions import SoftTimeLimitExceeded
-    from superset.reports.commands.exceptions import AlertQueryTimeout
 
     screenshot_mock.side_effect = SoftTimeLimitExceeded()
     with pytest.raises(ReportScheduleScreenshotTimeout):
         AsyncExecuteReportScheduleCommand(
-            test_id, create_alert_email_chart.id, datetime.utcnow()
+            TEST_ID, create_alert_email_chart.id, datetime.utcnow()
         ).run()
 
     # Assert the email smtp address, asserts a notification was sent with the error
-    assert email_mock.call_args[0][0] == "admin@fab.org"
+    assert email_mock.call_args[0][0] == OWNER_EMAIL
 
     assert_log(
         ReportState.ERROR, error_message="A timeout occurred while taking a screenshot."
     )
 
 
+@pytest.mark.usefixtures(
+    "load_birth_names_dashboard_with_slices", "create_report_email_chart_with_csv"
+)
+@patch("superset.utils.csv.urllib.request.urlopen")
+@patch("superset.utils.csv.urllib.request.OpenerDirector.open")
+@patch("superset.reports.notifications.email.send_email_smtp")
+@patch("superset.utils.csv.get_chart_csv_data")
+def test_soft_timeout_csv(
+    csv_mock, email_mock, mock_open, mock_urlopen, create_report_email_chart_with_csv,
+):
+    """
+    ExecuteReport Command: Test fail on generating csv
+    """
+    from celery.exceptions import SoftTimeLimitExceeded
+
+    response = Mock()
+    mock_open.return_value = response
+    mock_urlopen.return_value = response
+    mock_urlopen.return_value.getcode.side_effect = SoftTimeLimitExceeded()
+
+    with pytest.raises(ReportScheduleCsvTimeout):
+        AsyncExecuteReportScheduleCommand(
+            TEST_ID, create_report_email_chart_with_csv.id, datetime.utcnow()
+        ).run()
+
+    notification_targets = get_target_from_report_schedule(
+        create_report_email_chart_with_csv
+    )
+    # Assert the email smtp address, asserts a notification was sent with the error
+    assert email_mock.call_args[0][0] == OWNER_EMAIL
+
+    assert_log(
+        ReportState.ERROR, error_message="A timeout occurred while generating a csv.",
+    )
+
+
+@pytest.mark.usefixtures(
+    "load_birth_names_dashboard_with_slices", "create_report_email_chart_with_csv"
+)
+@patch("superset.utils.csv.urllib.request.urlopen")
+@patch("superset.utils.csv.urllib.request.OpenerDirector.open")
+@patch("superset.reports.notifications.email.send_email_smtp")
+@patch("superset.utils.csv.get_chart_csv_data")
+def test_generate_no_csv(
+    csv_mock, email_mock, mock_open, mock_urlopen, create_report_email_chart_with_csv,
+):
+    """
+    ExecuteReport Command: Test fail on generating csv
+    """
+    response = Mock()
+    mock_open.return_value = response
+    mock_urlopen.return_value = response
+    mock_urlopen.return_value.getcode.return_value = 200
+    response.read.return_value = None
+
+    with pytest.raises(ReportScheduleCsvFailedError):
+        AsyncExecuteReportScheduleCommand(
+            TEST_ID, create_report_email_chart_with_csv.id, datetime.utcnow()
+        ).run()
+
+    notification_targets = get_target_from_report_schedule(
+        create_report_email_chart_with_csv
+    )
+    # Assert the email smtp address, asserts a notification was sent with the error
+    assert email_mock.call_args[0][0] == OWNER_EMAIL
+
+    assert_log(
+        ReportState.ERROR,
+        error_message="Report Schedule execution failed when generating a csv.",
+    )
+
+
 @pytest.mark.usefixtures(
     "load_birth_names_dashboard_with_slices", "create_report_email_chart"
 )
@@ -946,23 +1134,59 @@ def test_fail_screenshot(screenshot_mock, email_mock, create_report_email_chart)
     ExecuteReport Command: Test soft timeout on screenshot
     """
     from celery.exceptions import SoftTimeLimitExceeded
+
     from superset.reports.commands.exceptions import AlertQueryTimeout
 
     screenshot_mock.side_effect = Exception("Unexpected error")
     with pytest.raises(ReportScheduleScreenshotFailedError):
         AsyncExecuteReportScheduleCommand(
-            test_id, create_report_email_chart.id, datetime.utcnow()
+            TEST_ID, create_report_email_chart.id, datetime.utcnow()
         ).run()
 
     notification_targets = get_target_from_report_schedule(create_report_email_chart)
     # Assert the email smtp address, asserts a notification was sent with the error
-    assert email_mock.call_args[0][0] == "admin@fab.org"
+    assert email_mock.call_args[0][0] == OWNER_EMAIL
 
     assert_log(
         ReportState.ERROR, error_message="Failed taking a screenshot Unexpected error"
     )
 
 
+@pytest.mark.usefixtures(
+    "load_birth_names_dashboard_with_slices", "create_report_email_chart_with_csv"
+)
+@patch("superset.reports.notifications.email.send_email_smtp")
+@patch("superset.utils.csv.urllib.request.urlopen")
+@patch("superset.utils.csv.urllib.request.OpenerDirector.open")
+@patch("superset.utils.csv.get_chart_csv_data")
+def test_fail_csv(
+    csv_mock, mock_open, mock_urlopen, email_mock, create_report_email_chart_with_csv
+):
+    """
+    ExecuteReport Command: Test error on csv
+    """
+
+    response = Mock()
+    mock_open.return_value = response
+    mock_urlopen.return_value = response
+    mock_urlopen.return_value.getcode.return_value = 500
+
+    with pytest.raises(ReportScheduleCsvFailedError):
+        AsyncExecuteReportScheduleCommand(
+            TEST_ID, create_report_email_chart_with_csv.id, datetime.utcnow()
+        ).run()
+
+    notification_targets = get_target_from_report_schedule(
+        create_report_email_chart_with_csv
+    )
+    # Assert the email smtp address, asserts a notification was sent with the error
+    assert email_mock.call_args[0][0] == OWNER_EMAIL
+
+    assert_log(
+        ReportState.ERROR, error_message="Failed generating csv <urlopen error 500>"
+    )
+
+
 @pytest.mark.usefixtures(
     "load_birth_names_dashboard_with_slices", "create_alert_email_chart"
 )
@@ -977,7 +1201,7 @@ def test_email_disable_screenshot(email_mock, create_alert_email_chart):
     """
 
     AsyncExecuteReportScheduleCommand(
-        test_id, create_alert_email_chart.id, datetime.utcnow()
+        TEST_ID, create_alert_email_chart.id, datetime.utcnow()
     ).run()
 
     notification_targets = get_target_from_report_schedule(create_alert_email_chart)
@@ -998,14 +1222,14 @@ def test_invalid_sql_alert(email_mock, create_invalid_sql_alert_email_chart):
     with freeze_time("2020-01-01T00:00:00Z"):
         with pytest.raises((AlertQueryError, AlertQueryInvalidTypeError)):
             AsyncExecuteReportScheduleCommand(
-                test_id, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
+                TEST_ID, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
             ).run()
 
         notification_targets = get_target_from_report_schedule(
             create_invalid_sql_alert_email_chart
         )
         # Assert the email smtp address, asserts a notification was sent with the error
-        assert email_mock.call_args[0][0] == "admin@fab.org"
+        assert email_mock.call_args[0][0] == OWNER_EMAIL
 
 
 @pytest.mark.usefixtures("create_invalid_sql_alert_email_chart")
@@ -1017,7 +1241,7 @@ def test_grace_period_error(email_mock, create_invalid_sql_alert_email_chart):
     with freeze_time("2020-01-01T00:00:00Z"):
         with pytest.raises((AlertQueryError, AlertQueryInvalidTypeError)):
             AsyncExecuteReportScheduleCommand(
-                test_id, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
+                TEST_ID, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
             ).run()
 
         # Only needed for MySQL, understand why
@@ -1026,7 +1250,7 @@ def test_grace_period_error(email_mock, create_invalid_sql_alert_email_chart):
             create_invalid_sql_alert_email_chart
         )
         # Assert the email smtp address, asserts a notification was sent with the error
-        assert email_mock.call_args[0][0] == "admin@fab.org"
+        assert email_mock.call_args[0][0] == OWNER_EMAIL
         assert (
             get_notification_error_sent_count(create_invalid_sql_alert_email_chart) == 1
         )
@@ -1034,7 +1258,7 @@ def test_grace_period_error(email_mock, create_invalid_sql_alert_email_chart):
     with freeze_time("2020-01-01T00:30:00Z"):
         with pytest.raises((AlertQueryError, AlertQueryInvalidTypeError)):
             AsyncExecuteReportScheduleCommand(
-                test_id, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
+                TEST_ID, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
             ).run()
         db.session.commit()
         assert (
@@ -1045,7 +1269,7 @@ def test_grace_period_error(email_mock, create_invalid_sql_alert_email_chart):
     with freeze_time("2020-01-01T01:30:00Z"):
         with pytest.raises((AlertQueryError, AlertQueryInvalidTypeError)):
             AsyncExecuteReportScheduleCommand(
-                test_id, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
+                TEST_ID, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
             ).run()
         db.session.commit()
         assert (
@@ -1057,7 +1281,7 @@ def test_grace_period_error(email_mock, create_invalid_sql_alert_email_chart):
 @patch("superset.reports.notifications.email.send_email_smtp")
 @patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
 def test_grace_period_error_flap(
-    screenshot_mock, email_mock, create_invalid_sql_alert_email_chart
+    screenshot_mock, email_mock, create_invalid_sql_alert_email_chart,
 ):
     """
     ExecuteReport Command: Test alert grace period on error
@@ -1065,7 +1289,7 @@ def test_grace_period_error_flap(
     with freeze_time("2020-01-01T00:00:00Z"):
         with pytest.raises((AlertQueryError, AlertQueryInvalidTypeError)):
             AsyncExecuteReportScheduleCommand(
-                test_id, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
+                TEST_ID, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
             ).run()
         db.session.commit()
         # Assert we have 1 notification sent on the log
@@ -1076,7 +1300,7 @@ def test_grace_period_error_flap(
     with freeze_time("2020-01-01T00:30:00Z"):
         with pytest.raises((AlertQueryError, AlertQueryInvalidTypeError)):
             AsyncExecuteReportScheduleCommand(
-                test_id, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
+                TEST_ID, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
             ).run()
         db.session.commit()
         assert (
@@ -1092,11 +1316,11 @@ def test_grace_period_error_flap(
     with freeze_time("2020-01-01T00:31:00Z"):
         # One success
         AsyncExecuteReportScheduleCommand(
-            test_id, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
+            TEST_ID, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
         ).run()
         # Grace period ends
         AsyncExecuteReportScheduleCommand(
-            test_id, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
+            TEST_ID, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
         ).run()
 
         db.session.commit()
@@ -1111,7 +1335,7 @@ def test_grace_period_error_flap(
     with freeze_time("2020-01-01T00:32:00Z"):
         with pytest.raises((AlertQueryError, AlertQueryInvalidTypeError)):
             AsyncExecuteReportScheduleCommand(
-                test_id, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
+                TEST_ID, create_invalid_sql_alert_email_chart.id, datetime.utcnow()
             ).run()
         db.session.commit()
         assert (
diff --git a/tests/reports/utils.py b/tests/reports/utils.py
index 2e92e1531..1c1f08ac6 100644
--- a/tests/reports/utils.py
+++ b/tests/reports/utils.py
@@ -23,6 +23,7 @@ from superset import db
 from superset.models.core import Database
 from superset.models.dashboard import Dashboard
 from superset.models.reports import (
+    ReportDataFormat,
     ReportExecutionLog,
     ReportRecipients,
     ReportSchedule,
@@ -47,6 +48,7 @@ def insert_report_schedule(
     last_state: Optional[ReportState] = None,
     grace_period: Optional[int] = None,
     recipients: Optional[List[ReportRecipients]] = None,
+    report_format: Optional[ReportDataFormat] = None,
     logs: Optional[List[ReportExecutionLog]] = None,
 ) -> ReportSchedule:
     owners = owners or []
@@ -70,6 +72,7 @@ def insert_report_schedule(
         recipients=recipients,
         logs=logs,
         last_state=last_state,
+        report_format=report_format,
     )
     db.session.add(report_schedule)
     db.session.commit()