feat: add application states

faststream/_internal/application.py

@@ -16,6 +16,7 @@
 from typing_extensions import ParamSpec
 
 from faststream.asyncapi.proto import AsyncAPIApplication
+from faststream.constants import AppState
 from faststream.log.logging import logger
 from faststream.utils import apply_types, context
 from faststream.utils.functions import drop_response_type, fake_context, to_async
@@ -70,6 +71,7 @@ def __init__(
     ) -> None:
         context.set_global("app", self)
 
+        self._state: AppState = AppState.STATE_STOPPED
         self._should_exit = False
         self.broker = broker
         self.logger = logger
@@ -197,15 +199,21 @@ async def _startup(
     ) -> None:
         self._log(log_level, "FastStream app starting...")
         await self.start(**(run_extra_options or {}))
+        self._state = AppState.STATE_RUNNING
         self._log(
             log_level, "FastStream app started successfully! To exit, press CTRL+C"
         )
 
     async def _shutdown(self, log_level: int = logging.INFO) -> None:
         self._log(log_level, "FastStream app shutting down...")
         await self.stop()
+        self._state = AppState.STATE_STOPPED
         self._log(log_level, "FastStream app shut down gracefully.")
 
     def _log(self, level: int, message: str) -> None:
         if self.logger is not None:
             self.logger.log(level, message)
+
+    @property
+    def state(self) -> AppState:
+        return self._state

faststream/constants.py

@@ -1,4 +1,4 @@
-from enum import Enum
+from enum import Enum, auto
 
 ContentType = str
 
@@ -8,3 +8,10 @@ class ContentTypes(str, Enum):
 
     text = "text/plain"
     json = "application/json"
+
+
+class AppState(str, Enum):
+    """Class with application states."""
+
+    STATE_STOPPED = auto()
+    STATE_RUNNING = auto()

tests/cli/test_app_state.py

@@ -0,0 +1,30 @@
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from faststream import FastStream
+from faststream.constants import AppState
+
+
+@pytest.mark.asyncio
+async def test_state_running(app: FastStream):
+    with patch(
+        "faststream._internal.application.Application.start", new_callable=AsyncMock
+    ):
+        await app._startup()
+        assert app.state == AppState.STATE_RUNNING
+
+
+@pytest.mark.asyncio
+async def test_state_stopped(app: FastStream):
+    with (
+        patch(
+            "faststream._internal.application.Application.start", new_callable=AsyncMock
+        ),
+        patch(
+            "faststream._internal.application.Application.stop", new_callable=AsyncMock
+        ),
+    ):
+        await app._startup()
+        await app._shutdown()
+        assert app.state == AppState.STATE_STOPPED