This is an automated email from the ASF dual-hosted git repository.
jli pushed a commit to branch 4.1
in repository https://gitbox.apache.org/repos/asf/superset.git
The following commit(s) were added to refs/heads/4.1 by this push:
new dc24c4f61e fix: Signature of Celery pruner jobs (#32699)
dc24c4f61e is described below
commit dc24c4f61e51887c6576d15eacc9865c648826bc
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 | 21 ++++++++++++++-------
3 files changed, 24 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 87bc0d28ad..17f5ec7ceb 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)",
- 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 ee3c1daea9..4b4c14f75d 100644
--- a/superset/tasks/scheduler.py
+++ b/superset/tasks/scheduler.py
@@ -14,10 +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
+from datetime import datetime, timezone
+from typing import Any
-from celery import Celery
+from celery import Task
from celery.exceptions import SoftTimeLimitExceeded
from superset import app, is_feature_enabled
@@ -76,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")
@@ -123,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() -> 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")
@@ -136,8 +141,10 @@ def prune_query() -> 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")
