willbarrett commented on a change in pull request #11711: URL: https://github.com/apache/incubator-superset/pull/11711#discussion_r524878020
########## File path: superset/reports/commands/alert.py ########## @@ -0,0 +1,79 @@ +# 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 json +import logging +from operator import eq, ge, gt, le, lt, ne +from typing import Optional + +import numpy as np + +from superset import jinja_context +from superset.commands.base import BaseCommand +from superset.models.reports import ReportSchedule, ReportScheduleValidatorType +from superset.reports.commands.exceptions import ( + AlertQueryInvalidTypeError, + AlertQueryMultipleColumnsError, + AlertQueryMultipleRowsError, +) + +logger = logging.getLogger(__name__) + + +OPERATOR_FUNCTIONS = {">=": ge, ">": gt, "<=": le, "<": lt, "==": eq, "!=": ne} + + +class AlertCommand(BaseCommand): + def __init__(self, report_schedule: ReportSchedule): + self._report_schedule = report_schedule + self._result: Optional[float] = None + + def run(self) -> bool: + self.validate() + self._report_schedule.last_value = self._result + + if self._report_schedule.validator_type == ReportScheduleValidatorType.NOT_NULL: + return self._result not in (0, None, np.nan) + operator = json.loads(self._report_schedule.validator_config_json)["op"] + threshold = json.loads(self._report_schedule.validator_config_json)["threshold"] + return OPERATOR_FUNCTIONS[operator](self._result, threshold) + + def validate(self) -> None: + """ + Validate the query result as a Pandas DataFrame + """ + sql_template = jinja_context.get_template_processor( + database=self._report_schedule.database + ) + rendered_sql = sql_template.process_template(self._report_schedule.sql) + df = self._report_schedule.database.get_df(rendered_sql) + + if df.empty: + return + rows = df.to_records() + # check if query return more then one row + if len(rows) > 1: + raise AlertQueryMultipleRowsError() + if len(rows[0]) > 2: + raise AlertQueryMultipleColumnsError() Review comment: I think we only want to raise here if the validator is something other than "NOT NULL" - if a row is returned, then it's not null :) ########## File path: superset/reports/commands/alert.py ########## @@ -0,0 +1,79 @@ +# 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 json +import logging +from operator import eq, ge, gt, le, lt, ne +from typing import Optional + +import numpy as np + +from superset import jinja_context +from superset.commands.base import BaseCommand +from superset.models.reports import ReportSchedule, ReportScheduleValidatorType +from superset.reports.commands.exceptions import ( + AlertQueryInvalidTypeError, + AlertQueryMultipleColumnsError, + AlertQueryMultipleRowsError, +) + +logger = logging.getLogger(__name__) + + +OPERATOR_FUNCTIONS = {">=": ge, ">": gt, "<=": le, "<": lt, "==": eq, "!=": ne} + + +class AlertCommand(BaseCommand): + def __init__(self, report_schedule: ReportSchedule): + self._report_schedule = report_schedule + self._result: Optional[float] = None + + def run(self) -> bool: + self.validate() + self._report_schedule.last_value = self._result + + if self._report_schedule.validator_type == ReportScheduleValidatorType.NOT_NULL: + return self._result not in (0, None, np.nan) Review comment: I think this is correct if the result is a float, but if it's a row or set of rows this should behave differently. ########## File path: superset/reports/commands/exceptions.py ########## @@ -103,10 +103,42 @@ class ReportScheduleDeleteFailedError(CommandException): message = _("Report Schedule delete failed.") +class PruneReportScheduleLogFailedError(CommandException): + message = _("Report Schedule log prune failed.") + + +class ReportScheduleScreenshotFailedError(CommandException): + message = _("Report Schedule execute screenshot failed.") + + +class ReportScheduleExecuteUnexpectedError(CommandException): + message = _("Report Schedule execute got an unexpected error.") + + +class ReportSchedulePreviousWorkingError(CommandException): + message = _("Report Schedule is still working error, refused to re compute.") Review comment: Nit, English: "Report Schedule is still working, refusing to re-compute" ########## File path: superset/reports/commands/exceptions.py ########## @@ -103,10 +103,42 @@ class ReportScheduleDeleteFailedError(CommandException): message = _("Report Schedule delete failed.") +class PruneReportScheduleLogFailedError(CommandException): + message = _("Report Schedule log prune failed.") + + +class ReportScheduleScreenshotFailedError(CommandException): + message = _("Report Schedule execute screenshot failed.") Review comment: Nit, English: "Report Schedule execution failed when generating a screenshot" ########## File path: superset/reports/commands/execute.py ########## @@ -0,0 +1,229 @@ +# 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 logging +from datetime import datetime, timedelta +from typing import Optional + +from sqlalchemy.orm import Session + +from superset import app, thumbnail_cache +from superset.commands.base import BaseCommand +from superset.commands.exceptions import CommandException +from superset.extensions import security_manager +from superset.models.reports import ( + ReportExecutionLog, + ReportLogState, + ReportSchedule, + ReportScheduleType, +) +from superset.reports.commands.alert import AlertCommand +from superset.reports.commands.exceptions import ( + ReportScheduleAlertGracePeriodError, + ReportScheduleExecuteUnexpectedError, + ReportScheduleNotFoundError, + ReportSchedulePreviousWorkingError, + ReportScheduleScreenshotFailedError, +) +from superset.reports.dao import ReportScheduleDAO +from superset.reports.notifications import create_notification +from superset.reports.notifications.base import NotificationContent, ScreenshotData +from superset.utils.celery import session_scope +from superset.utils.screenshots import ( + BaseScreenshot, + ChartScreenshot, + DashboardScreenshot, +) +from superset.utils.urls import get_url_path + +logger = logging.getLogger(__name__) + + +class AsyncExecuteReportScheduleCommand(BaseCommand): Review comment: I read this class a few times looking for suggestions on simplification or breaking it up. I didn't come up with much, but I figured I'd surface that I felt the need to make the effort. ########## File path: superset/reports/notifications/__init__.py ########## @@ -0,0 +1,34 @@ +# -*- coding: utf-8 -*- +# 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. +from superset.models.reports import ReportRecipients +from superset.reports.notifications.base import BaseNotification, NotificationContent +from superset.reports.notifications.email import EmailNotification +from superset.reports.notifications.slack import SlackNotification + + +def create_notification( Review comment: This is the rare instance where I support a non-empty `__init__.py`. Revel, Daniel, in your success. ########## File path: superset/reports/commands/exceptions.py ########## @@ -103,10 +103,42 @@ class ReportScheduleDeleteFailedError(CommandException): message = _("Report Schedule delete failed.") +class PruneReportScheduleLogFailedError(CommandException): + message = _("Report Schedule log prune failed.") + + +class ReportScheduleScreenshotFailedError(CommandException): + message = _("Report Schedule execute screenshot failed.") + + +class ReportScheduleExecuteUnexpectedError(CommandException): + message = _("Report Schedule execute got an unexpected error.") Review comment: Nit, English: "There was an unexpected error in Report Schedule execution" ########## File path: superset/reports/commands/exceptions.py ########## @@ -103,10 +103,42 @@ class ReportScheduleDeleteFailedError(CommandException): message = _("Report Schedule delete failed.") +class PruneReportScheduleLogFailedError(CommandException): + message = _("Report Schedule log prune failed.") + + +class ReportScheduleScreenshotFailedError(CommandException): + message = _("Report Schedule execute screenshot failed.") + + +class ReportScheduleExecuteUnexpectedError(CommandException): + message = _("Report Schedule execute got an unexpected error.") + + +class ReportSchedulePreviousWorkingError(CommandException): + message = _("Report Schedule is still working error, refused to re compute.") + + class ReportScheduleNameUniquenessValidationError(ValidationError): """ Marshmallow validation error for Report Schedule name already exists """ def __init__(self) -> None: super().__init__([_("Name must be unique")], field_name="name") + + +class AlertQueryMultipleRowsError(CommandException): + message = _("Alert query returned more then one row.") + + +class AlertQueryMultipleColumnsError(CommandException): + message = _("Alert query returned more then one column.") + + +class AlertQueryInvalidTypeError(CommandException): + message = _("Alert query returned a non-number value") + + +class ReportScheduleAlertGracePeriodError(CommandException): + message = _("Alert on grace period") Review comment: Nit, English: "Alert fired during grace period" ########## File path: superset/reports/commands/exceptions.py ########## @@ -103,10 +103,42 @@ class ReportScheduleDeleteFailedError(CommandException): message = _("Report Schedule delete failed.") +class PruneReportScheduleLogFailedError(CommandException): + message = _("Report Schedule log prune failed.") + + +class ReportScheduleScreenshotFailedError(CommandException): + message = _("Report Schedule execute screenshot failed.") + + +class ReportScheduleExecuteUnexpectedError(CommandException): + message = _("Report Schedule execute got an unexpected error.") + + +class ReportSchedulePreviousWorkingError(CommandException): + message = _("Report Schedule is still working error, refused to re compute.") + + class ReportScheduleNameUniquenessValidationError(ValidationError): """ Marshmallow validation error for Report Schedule name already exists """ def __init__(self) -> None: super().__init__([_("Name must be unique")], field_name="name") + + +class AlertQueryMultipleRowsError(CommandException): + message = _("Alert query returned more then one row.") + + +class AlertQueryMultipleColumnsError(CommandException): + message = _("Alert query returned more then one column.") + + +class AlertQueryInvalidTypeError(CommandException): + message = _("Alert query returned a non-number value") Review comment: Nit, English: "Alert query returned a non-numeric value" ########## File path: superset/reports/commands/alert.py ########## @@ -0,0 +1,79 @@ +# 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 json +import logging +from operator import eq, ge, gt, le, lt, ne +from typing import Optional + +import numpy as np + +from superset import jinja_context +from superset.commands.base import BaseCommand +from superset.models.reports import ReportSchedule, ReportScheduleValidatorType +from superset.reports.commands.exceptions import ( + AlertQueryInvalidTypeError, + AlertQueryMultipleColumnsError, + AlertQueryMultipleRowsError, +) + +logger = logging.getLogger(__name__) + + +OPERATOR_FUNCTIONS = {">=": ge, ">": gt, "<=": le, "<": lt, "==": eq, "!=": ne} + + +class AlertCommand(BaseCommand): + def __init__(self, report_schedule: ReportSchedule): + self._report_schedule = report_schedule + self._result: Optional[float] = None + + def run(self) -> bool: + self.validate() + self._report_schedule.last_value = self._result + + if self._report_schedule.validator_type == ReportScheduleValidatorType.NOT_NULL: + return self._result not in (0, None, np.nan) + operator = json.loads(self._report_schedule.validator_config_json)["op"] + threshold = json.loads(self._report_schedule.validator_config_json)["threshold"] + return OPERATOR_FUNCTIONS[operator](self._result, threshold) + + def validate(self) -> None: + """ + Validate the query result as a Pandas DataFrame + """ + sql_template = jinja_context.get_template_processor( + database=self._report_schedule.database + ) + rendered_sql = sql_template.process_template(self._report_schedule.sql) + df = self._report_schedule.database.get_df(rendered_sql) + + if df.empty: + return + rows = df.to_records() + # check if query return more then one row + if len(rows) > 1: + raise AlertQueryMultipleRowsError() Review comment: I think we only want to raise here if the validator is something other than "NOT NULL" ########## File path: superset/reports/notifications/email.py ########## @@ -0,0 +1,94 @@ +# -*- coding: utf-8 -*- +# 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 json +import logging +from dataclasses import dataclass +from email.utils import make_msgid, parseaddr +from typing import Dict + +from flask_babel import gettext as __ + +from superset import app +from superset.models.reports import ReportRecipientType +from superset.reports.notifications.base import BaseNotification +from superset.utils.core import send_email_smtp + +logger = logging.getLogger(__name__) + + +@dataclass +class EmailContent: + body: str + images: Dict[str, bytes] + + +class EmailNotification(BaseNotification): # pylint: disable=too-few-public-methods + """ + Sends an email notification for a report recipient + """ + + type = ReportRecipientType.EMAIL + + @staticmethod + def _get_smtp_domain() -> str: + return parseaddr(app.config["SMTP_MAIL_FROM"])[1].split("@")[1] + + def _get_content(self) -> EmailContent: + # Get the domain from the 'From' address .. + # and make a message id without the < > in the ends + domain = self._get_smtp_domain() + msgid = make_msgid(domain)[1:-1] + + image = {msgid: self._content.screenshot.image} + body = __( + """ + <b><a href="%(url)s">Explore in Superset</a></b><p></p> + <img src="cid:%(msgid)s"> + """, + url=self._content.screenshot.url, + msgid=msgid, + ) + return EmailContent(body=body, images=image) + + def _get_subject(self) -> str: + return __( + "%(prefix)s %(title)s", + prefix=app.config["EMAIL_REPORTS_SUBJECT_PREFIX"], + title=self._content.name, + ) + + def _get_to(self) -> str: + return json.loads(self._recipient.recipient_config_json)["target"] + + def send(self) -> None: + subject = self._get_subject() + content = self._get_content() + to = self._get_to() + send_email_smtp( + to, Review comment: This is an interesting question - should we treat this as a single SMTP send with all of the recipients in the `to` address field, or split recipients by `,` and send to each one individually? My preference would be the latter, but open to discussion. Eventually we should provide configuration to allow MTA batching. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
