fix: add explicit type annotations to new datetime locals

Addresses codeant-ai bot review threads asking for explicit type hints
on the datetime variables introduced by the utcnow() replacement.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

superset/commands/report/execute.py

@@ -134,7 +134,7 @@ class BaseReportState:
     ) -> None:
         self._report_schedule = report_schedule
         self._scheduled_dttm = scheduled_dttm
-        self._start_dttm = datetime.now(timezone.utc).replace(tzinfo=None)
+        self._start_dttm: datetime = datetime.now(timezone.utc).replace(tzinfo=None)
         self._execution_id = execution_id
         self._filter_warnings: list[str] = []
 
@@ -513,7 +513,7 @@ class BaseReportState:
         Get chart or dashboard screenshots
         :raises: ReportScheduleScreenshotFailedError
         """
-        start_time = datetime.now(timezone.utc).replace(tzinfo=None)
+        start_time: datetime = datetime.now(timezone.utc).replace(tzinfo=None)
 
         user, _ = resolve_executor_user(self._report_schedule)
 
@@ -707,7 +707,7 @@ class BaseReportState:
         return content or None
 
     def _get_csv_data(self) -> bytes:
-        start_time = datetime.now(timezone.utc).replace(tzinfo=None)
+        start_time: datetime = datetime.now(timezone.utc).replace(tzinfo=None)
         user, username = resolve_executor_user(self._report_schedule)
         auth_cookies = machine_auth_provider_factory.instance.get_auth_cookies(user)
 
@@ -775,7 +775,7 @@ class BaseReportState:
         """
         Return data as a Pandas dataframe, to embed in notifications as a table.
         """
-        start_time = datetime.now(timezone.utc).replace(tzinfo=None)
+        start_time: datetime = datetime.now(timezone.utc).replace(tzinfo=None)
 
         url = self._get_url(result_format=ChartDataResultFormat.JSON)
         user, username = resolve_executor_user(self._report_schedule)
@@ -1413,7 +1413,7 @@ class AsyncExecuteReportScheduleCommand(BaseCommand):
             )
             user = security_manager.find_user(username)
 
-            start_time = datetime.now(timezone.utc).replace(tzinfo=None)
+            start_time: datetime = datetime.now(timezone.utc).replace(tzinfo=None)
             with override_user(user):
                 # Pre-commit any permalink rows before the state machine's
                 # @transaction() opens. When called inside a transaction,

superset/commands/report/log_prune.py

@@ -41,9 +41,9 @@ class AsyncPruneReportScheduleLogCommand(BaseCommand):
 
         for report_schedule in db.session.query(ReportSchedule).all():
             if report_schedule.log_retention is not None:
-                from_date = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(
-                    days=report_schedule.log_retention
-                )
+                from_date: datetime = datetime.now(timezone.utc).replace(
+                    tzinfo=None
+                ) - timedelta(days=report_schedule.log_retention)
                 try:
                     row_count = ReportScheduleDAO.bulk_delete_logs(
                         report_schedule,

superset/daos/query.py

@@ -48,7 +48,7 @@ class QueryDAO(BaseDAO[Query]):
     @staticmethod
     def get_queries_changed_after(last_updated_ms: Union[float, int]) -> list[Query]:
         # UTC date time, same that is stored in the DB.
-        last_updated_dt = datetime.fromtimestamp(
+        last_updated_dt: datetime = datetime.fromtimestamp(
             last_updated_ms / 1000, timezone.utc
         ).replace(tzinfo=None)
 

superset/utils/cache.py

@@ -226,7 +226,9 @@ def etag_cache(  # noqa: C901
 
             # Check if the cache is stale. Default the content_changed_time to now
             # if we don't know when it was last modified.
-            content_changed_time = datetime.now(timezone.utc).replace(tzinfo=None)
+            content_changed_time: datetime = datetime.now(timezone.utc).replace(
+                tzinfo=None
+            )
             if get_last_modified:
                 content_changed_time = get_last_modified(*args, **kwargs)
                 if (