Correctly handle None returns from Query.scalar() (#16345)

This is possible when the query does not return a row, according to
SQLAlchemy documentation. We can handle them to provide better errors in
unexpected situations.

Toward #8171, fix #16328.

airflow/models/serialized_dag.py

@@ -280,7 +280,7 @@ def get_last_updated_datetime(cls, dag_id: str, session: Session = None) -> Opti
 
     @classmethod
     @provide_session
-    def get_max_last_updated_datetime(cls, session: Session = None) -> datetime:
+    def get_max_last_updated_datetime(cls, session: Session = None) -> Optional[datetime]:
         """
         Get the maximum date when any DAG was last updated in serialized_dag table
 
@@ -291,15 +291,15 @@ def get_max_last_updated_datetime(cls, session: Session = None) -> datetime:
 
     @classmethod
     @provide_session
-    def get_latest_version_hash(cls, dag_id: str, session: Session = None) -> str:
+    def get_latest_version_hash(cls, dag_id: str, session: Session = None) -> Optional[str]:
         """
         Get the latest DAG version for a given DAG ID.
 
         :param dag_id: DAG ID
         :type dag_id: str
         :param session: ORM Session
         :type session: Session
-        :return: DAG Hash
+        :return: DAG Hash, or None if the DAG is not found
         :rtype: str | None
         """
         return session.query(cls.dag_hash).filter(cls.dag_id == dag_id).scalar()

airflow/www/views.py

@@ -4093,7 +4093,8 @@ def list(self):
         title = "DAG Dependencies"
 
         if timezone.utcnow() > self.last_refresh + self.refresh_interval:
-            if SerializedDagModel.get_max_last_updated_datetime() > self.last_refresh:
+            max_last_updated = SerializedDagModel.get_max_last_updated_datetime()
+            if max_last_updated is None or max_last_updated > self.last_refresh:
                 self._calculate_graph()
             self.last_refresh = timezone.utcnow()