fix: Signature of Celery pruner jobs (#32699)

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:

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:

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")