Vitor-Avila commented on code in PR #44992:
URL: https://github.com/apache/superset/pull/44992#discussion_r4195397484
##########
superset/daos/report.py:
##########
@@ -392,3 +399,215 @@ def bulk_delete_logs(model: ReportSchedule, from_date:
datetime) -> int | None:
)
.delete(synchronize_session="fetch")
)
+
+ @staticmethod
+ def invalidate_pending_executions(schedule_ids: list[int] | None = None)
-> None:
+ """Fence queued retries and running attempts after a configuration
save.
+
+ The caller's transaction holds row locks only while saving metadata.
+ Preserve execution windows so old cron messages cannot replay the work.
+ """
+ query = db.session.query(ReportSchedule).filter(
+ ReportSchedule.last_state.in_([ReportState.WORKING,
ReportState.RETRYING])
+ )
+ if schedule_ids is not None:
+ query = query.filter(ReportSchedule.id.in_(schedule_ids))
+ for schedule in query.populate_existing().with_for_update().all():
+ if schedule.execution_owner:
+ db.session.query(ReportExecutionLog).filter(
+ ReportExecutionLog.report_schedule_id == schedule.id,
+ ReportExecutionLog.uuid == UUID(schedule.execution_owner),
+ ReportExecutionLog.state.in_(
+ [ReportState.WORKING, ReportState.RETRYING]
+ ),
+ ).update(
+ {
+ ReportExecutionLog.state: ReportState.ERROR,
+ ReportExecutionLog.end_dttm: datetime.utcnow(),
+ ReportExecutionLog.error_message: (
+ "Execution cancelled: configuration changed"
+ ),
+ },
+ synchronize_session=False,
+ )
+ schedule.execution_owner = str(uuid4())
+ schedule.last_state = ReportState.ERROR
+ schedule.last_eval_dttm = datetime.utcnow()
+ schedule.retry_attempt = 0
+ schedule.retry_scheduled_dttm = None
+
+ @staticmethod
+ def find_with_email_recipients() -> list[ReportSchedule]:
+ """
+ Find every schedule (active or not) with at least one e-mail recipient.
+ """
+ return (
+ db.session.query(ReportSchedule)
+ .join(
+ ReportRecipients,
+ ReportRecipients.report_schedule_id == ReportSchedule.id,
+ )
+ .filter(ReportRecipients.type == ReportRecipientType.EMAIL)
+ .options(selectinload(ReportSchedule.recipients))
+ .distinct()
+ .all()
+ )
+
+ @staticmethod
+ def find_by_type(report_type: ReportScheduleType) -> list[ReportSchedule]:
+ """
+ Find every schedule (active or not) of the given type.
+ """
+ return (
+ db.session.query(ReportSchedule)
+ .filter(ReportSchedule.type == report_type)
+ .all()
+ )
+
+
+class ReportConfigDAO:
+ """
+ Access to the global Alerts & Reports configuration (``report_config``).
+
+ Values are stored as JSON-encoded key-value rows. A key without a row is
+ "not configured" and resolves to the legacy application config or feature
+ flag through :meth:`get_effective_value`, which keeps deployments that
never
+ open the configuration UI behaving exactly as before.
+ """
+
+ @staticmethod
+ def get_stored_values() -> dict[str, Any]:
+ """
+ Return the raw stored configuration, keyed by :class:`ReportConfigKey`.
+
+ Read and decoding failures propagate: unavailable policy is not an
+ unrestricted policy. A stored null remains distinct from a missing row.
+ """
+ rows = db.session.query(ReportConfig).all()
+ return {
+ row.key: json.loads(row.value) if row.value is not None else None
+ for row in rows
+ }
+
+ @staticmethod
+ def get_fallback_value(key: ReportConfigKey) -> Any:
+ """Return the legacy application config / feature flag value for
``key``."""
+ if key == ReportConfigKey.ALERTS_ATTACH_REPORTS:
+ return
feature_flag_manager.is_feature_enabled("ALERTS_ATTACH_REPORTS")
+ if key == ReportConfigKey.DATE_FORMAT_IN_EMAIL_SUBJECT:
+ return feature_flag_manager.is_feature_enabled(
+ "DATE_FORMAT_IN_EMAIL_SUBJECT"
+ )
+ if key in (
+ ReportConfigKey.ALERT_MINIMUM_INTERVAL,
+ ReportConfigKey.REPORT_MINIMUM_INTERVAL,
+ ):
+ value = current_app.config.get(key.upper(), 0)
+ return value() if callable(value) else value
+ if key == ReportConfigKey.LIMIT_RECIPIENTS_TO_USERS:
+ return False
+ if key == ReportConfigKey.ALLOWED_EMAIL_DOMAINS:
+ return []
+ return None
+
+ @staticmethod
+ def get_effective_value(key: ReportConfigKey) -> Any:
+ """
+ Return the value in effect for ``key``: the stored value when present,
+ otherwise the legacy fallback.
+ """
+ if key in (stored := ReportConfigDAO.get_stored_values()):
+ return stored[key]
+ return ReportConfigDAO.get_fallback_value(key)
+
+ @staticmethod
+ def get_effective_config() -> dict[str, Any]:
+ """Return the value in effect for every :class:`ReportConfigKey`."""
+ stored = ReportConfigDAO.get_stored_values()
+ return {
+ key.value: (
+ stored[key]
+ if key in stored
+ else ReportConfigDAO.get_fallback_value(key)
+ )
+ for key in ReportConfigKey
+ }
+
+ @staticmethod
+ def upsert(values: dict[str, Any]) -> None:
+ """
+ Store values, retaining an explicit null as a configured value.
+ Only keys without a row inherit application configuration. Does not
commit.
+ """
+ rows = {row.key: row for row in db.session.query(ReportConfig).all()}
+ for key, value in values.items():
+ key = str(ReportConfigKey(key))
+ row = rows.get(key)
+ if row is None:
+ row = ReportConfig(key=key)
+ db.session.add(row)
Review Comment:
I think that's OK. The unique `key` constraint should fail the second
attempt, which can then be retried.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
