fix(reports): handle exceptions properly in scope (#12811)

* fix(reports): handle exceptions properly in scope

* list and test commit removal

* revert removing commits from tests

* lint

superset/tasks/scheduler.py

@@ -19,6 +19,7 @@
 from typing import Iterator
 
 import croniter
+from dateutil import parser
 
 from superset import app
 from superset.commands.exceptions import CommandException
@@ -60,9 +61,10 @@ def scheduler() -> None:
 
 
 @celery_app.task(name="reports.execute")
-def execute(report_schedule_id: int, scheduled_dttm: datetime) -> None:
+def execute(report_schedule_id: int, scheduled_dttm: str) -> None:
     try:
-        AsyncExecuteReportScheduleCommand(report_schedule_id, scheduled_dttm).run()
+        scheduled_dttm_ = parser.parse(scheduled_dttm)
+        AsyncExecuteReportScheduleCommand(report_schedule_id, scheduled_dttm_).run()
     except ReportScheduleUnexpectedError as ex:
         logger.error("An unexpected occurred while executing the report: %s", ex)
     except CommandException as ex:

superset/utils/celery.py

@@ -17,8 +17,9 @@
 import logging
 from typing import Iterator
 
-import sqlalchemy as sa
 from contextlib2 import contextmanager
+from sqlalchemy import create_engine
+from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import Session, sessionmaker
 from sqlalchemy.pool import NullPool
 
@@ -38,7 +39,7 @@ def session_scope(nullpool: bool) -> Iterator[Session]:
             in a future version of Superset."
         )
     if nullpool:
-        engine = sa.create_engine(database_uri, poolclass=NullPool)
+        engine = create_engine(database_uri, poolclass=NullPool)
         session_class = sessionmaker()
         session_class.configure(bind=engine)
         session = session_class()
@@ -49,7 +50,7 @@ def session_scope(nullpool: bool) -> Iterator[Session]:
     try:
         yield session
         session.commit()
-    except Exception as ex:
+    except SQLAlchemyError as ex:
         session.rollback()
         logger.exception(ex)
         raise