lightning apps: add flow fail() (#18883)

* add flow fail()

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
(cherry picked from commit fd48627)

src/lightning/app/core/app.py

@@ -51,7 +51,7 @@
 from lightning.app.utilities.commands.base import _process_requests
 from lightning.app.utilities.component import _convert_paths_after_init, _validate_root_flow
 from lightning.app.utilities.enum import AppStage, CacheCallsKeys
-from lightning.app.utilities.exceptions import CacheMissException, ExitAppException
+from lightning.app.utilities.exceptions import CacheMissException, ExitAppException, LightningFlowException
 from lightning.app.utilities.layout import _collect_layout
 from lightning.app.utilities.proxies import ComponentDelta
 from lightning.app.utilities.scheduler import SchedulerThread
@@ -462,6 +462,9 @@ def run_once(self) -> bool:
                 self.root.run()
         except CacheMissException:
             self._on_cache_miss_exception()
+        except LightningFlowException:
+            done = True
+            self.stage = AppStage.FAILED
         except (ExitAppException, KeyboardInterrupt):
             done = True
             self.stage = AppStage.STOPPING

src/lightning/app/core/flow.py

@@ -27,7 +27,7 @@
 from lightning.app.storage.path import Path
 from lightning.app.utilities.app_helpers import _is_json_serializable, _LightningAppRef, _set_child_name, is_overridden
 from lightning.app.utilities.component import _sanitize_state
-from lightning.app.utilities.exceptions import ExitAppException
+from lightning.app.utilities.exceptions import ExitAppException, LightningFlowException
 from lightning.app.utilities.introspection import _is_init_context, _is_run_context
 from lightning.app.utilities.packaging.cloud_compute import CloudCompute, _maybe_create_cloud_compute
 
@@ -391,6 +391,12 @@ def stop(self, end_msg: str = "") -> None:
             print(end_msg)
         raise ExitAppException
 
+    def fail(self, end_msg: str = "") -> None:
+        """Method used to exit and fail the application."""
+        if end_msg:
+            print(end_msg)
+        raise LightningFlowException
+
     def _exit(self, end_msg: str = "") -> None:
         """Used to exit the application.