This is an automated email from the ASF dual-hosted git repository. michaelsmolina pushed a commit to branch 5.0 in repository https://gitbox.apache.org/repos/asf/superset.git
commit d2ba0fc9aea30e6c031eb55b1a4f369822c70a60 Author: Michael S. Molina <[email protected]> AuthorDate: Mon Mar 17 13:52:31 2025 -0300 fix: Signature of Celery pruner jobs (#32699) (cherry picked from commit df06bdf33b460fbf9b37dbf2b20e14d1c490e414) --- superset/commands/logs/prune.py | 8 +++++--- superset/commands/sql_lab/query.py | 8 +++++--- superset/tasks/scheduler.py | 20 +++++++++++++------- 3 files changed, 23 insertions(+), 13 deletions(-) diff --git a/superset/commands/logs/prune.py b/superset/commands/logs/prune.py index 6a9cffb3f7..9ad031ea51 100644 --- a/superset/commands/logs/prune.py +++ b/superset/commands/logs/prune.py @@ -69,7 +69,7 @@ class LogPruneCommand(BaseCommand): total_rows = len(ids_to_delete) - logger.info("Total rows to be deleted: %s", total_rows) + logger.info("Total rows to be deleted: %s", f"{total_rows:,}") next_logging_threshold = 1 @@ -92,7 +92,7 @@ class LogPruneCommand(BaseCommand): if percentage_complete >= next_logging_threshold: logger.info( "Deleted %s rows from the logs table older than %s days (%d%% complete)", # noqa: E501 - total_deleted, + f"{total_deleted:,}", self.retention_period_days, percentage_complete, ) @@ -102,7 +102,9 @@ class LogPruneCommand(BaseCommand): minutes, seconds = divmod(elapsed_time, 60) formatted_time = f"{int(minutes):02}:{int(seconds):02}" logger.info( - "Pruning complete: %s rows deleted in %s", total_deleted, formatted_time + "Pruning complete: %s rows deleted in %s", + f"{total_deleted:,}", + formatted_time, ) def validate(self) -> None: diff --git a/superset/commands/sql_lab/query.py b/superset/commands/sql_lab/query.py index 466d5f99c9..b5263fc2e2 100644 --- a/superset/commands/sql_lab/query.py +++ b/superset/commands/sql_lab/query.py @@ -69,7 +69,7 @@ class QueryPruneCommand(BaseCommand): total_rows = len(ids_to_delete) - logger.info("Total rows to be deleted: %s", total_rows) + logger.info("Total rows to be deleted: %s", f"{total_rows:,}") next_logging_threshold = 1 @@ -92,7 +92,7 @@ class QueryPruneCommand(BaseCommand): if percentage_complete >= next_logging_threshold: logger.info( "Deleted %s rows from the query table older than %s days (%d%% complete)", # noqa: E501 - total_deleted, + f"{total_deleted:,}", self.retention_period_days, percentage_complete, ) @@ -102,7 +102,9 @@ class QueryPruneCommand(BaseCommand): minutes, seconds = divmod(elapsed_time, 60) formatted_time = f"{int(minutes):02}:{int(seconds):02}" logger.info( - "Pruning complete: %s rows deleted in %s", total_deleted, formatted_time + "Pruning complete: %s rows deleted in %s", + f"{total_deleted:,}", + formatted_time, ) def validate(self) -> None: diff --git a/superset/tasks/scheduler.py b/superset/tasks/scheduler.py index f894a6794b..3fec34d816 100644 --- a/superset/tasks/scheduler.py +++ b/superset/tasks/scheduler.py @@ -14,11 +14,13 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +from __future__ import annotations + import logging from datetime import datetime, timezone -from typing import Optional +from typing import Any -from celery import Celery +from celery import Task from celery.exceptions import SoftTimeLimitExceeded from superset import app, is_feature_enabled @@ -77,7 +79,7 @@ def scheduler() -> None: @celery_app.task(name="reports.execute", bind=True) -def execute(self: Celery.task, report_schedule_id: int) -> None: +def execute(self: Task, report_schedule_id: int) -> None: stats_logger: BaseStatsLogger = app.config["STATS_LOGGER"] stats_logger.incr("reports.execute") @@ -124,8 +126,10 @@ def prune_log() -> None: logger.exception("An exception occurred while pruning report schedule logs") -@celery_app.task(name="prune_query") -def prune_query(retention_period_days: Optional[int] = None) -> None: +@celery_app.task(name="prune_query", bind=True) +def prune_query( + self: Task, retention_period_days: int | None = None, **kwargs: Any +) -> None: stats_logger: BaseStatsLogger = app.config["STATS_LOGGER"] stats_logger.incr("prune_query") @@ -145,8 +149,10 @@ def prune_query(retention_period_days: Optional[int] = None) -> None: logger.exception("An error occurred while pruning queries: %s", ex) -@celery_app.task(name="prune_logs") -def prune_logs(retention_period_days: Optional[int] = None) -> None: +@celery_app.task(name="prune_logs", bind=True) +def prune_logs( + self: Task, retention_period_days: int | None = None, **kwargs: Any +) -> None: stats_logger: BaseStatsLogger = app.config["STATS_LOGGER"] stats_logger.incr("prune_logs")
