docs

docs/code_snippets/response/data_sharing.py

@@ -0,0 +1,49 @@
+from argenta import Router, Response, Command
+from argenta.command import Flag
+from argenta.data_bridge import DataBridge
+from argenta.di import FromDishka
+
+# 1. Создаём роутер
+router = Router(title='Authentication')
+
+# 2. Определяем сервис и обработчики
+def authenticate_user(username: str) -> str:
+    """Возвращает фиктивный токен для пользователя."""
+    return f"token_for_{username}"
+
+@router.command(Command('login', flags=Flag('username')))
+def login_handler(response: Response, data_bridge: FromDishka[DataBridge]):
+    """Обработчик для команды 'login'. Сохраняет токен в хранилище."""
+    username_flag = response.input_flags.get_flag_by_name('username')
+    if not username_flag or not username_flag.input_value:
+        print("[red]Ошибка:[/red] необходимо указать имя пользователя с помощью флага --username.")
+        return
+
+    username = username_flag.input_value
+    token = authenticate_user(username)
+
+    # Сохраняем токен в общем хранилище сессии
+    data_bridge.update({"auth_token": token})
+    print(f"[green]Успешный вход![/green] Пользователь '{username}' аутентифицирован.")
+
+@router.command('get-profile')
+def get_profile_handler(response: Response, data_bridge: FromDishka[DataBridge]):
+    """Обработчик для команды 'get-profile'. Использует токен из хранилища."""
+    session_data = data_bridge.get_all()
+    token = session_data.get("auth_token")
+
+    if not token:
+        print("[red]Ошибка:[/red] вы не аутентифицированы. Сначала выполните команду 'login'.")
+        return
+
+    print(f"Загрузка профиля с использованием токена: [yellow]{token}[/yellow]")
+
+@router.command('logout')
+def logout_handler(response: Response, data_bridge: FromDishka[DataBridge]):
+    """Обработчик для команды 'logout'. Очищает токен."""
+    try:
+        data_bridge.delete_by_key("auth_token")
+        print("[green]Выход выполнен.[/green] Данные сессии очищены.")
+    except KeyError:
+        print("Вы и так не были аутентифицированы.")
+

docs/root/api/bridge.rst

@@ -0,0 +1,60 @@
+.. _root_api_bridge:
+
+DataBridge
+==========
+
+`DataBridge` — это сущность, предоставляющая временное хранилище данных, которое существует в рамках одной сессии приложения (от запуска до выхода). Она предназначена для обмена данными между вызовами разных команд.
+
+Основной способ получения доступа к `DataBridge` — через систему внедрения зависимостей (DI).
+
+.. code-block:: python
+   :linenos:
+
+   from dishka.integrations.fastapi import FromDishka
+   from argenta.bridge import DataBridge
+
+   def my_handler(data_bridge: FromDishka[DataBridge]):
+       # ... ваш код
+
+**Практический пример: Аутентификация**
+
+Рассмотрим пример, где команда `login` сохраняет токен аутентификации, а команда `get-profile` использует его.
+
+.. literalinclude:: ../../code_snippets/response/data_sharing.py
+   :language: python
+   :linenos:
+
+**Как это работает:**
+
+1.  При вызове обработчика `dishka` автоматически внедряет экземпляр `DataBridge`.
+2.  Команда ``login --username <имя>`` вызывает `login_handler`, который через внедрённый `data_bridge` сохраняет токен.
+3.  Команда `get-profile` вызывает `get_profile_handler`, который так же получает `data_bridge` и извлекает из него токен.
+
+API класса
+-----------
+
+.. py:class:: DataBridge
+
+   .. py:method:: __init__(self, initial_data: dict | None = None)
+
+      Инициализирует хранилище. При использовании через DI вызывается автоматически.
+
+   .. py:method:: update(self, data: dict) -> None
+
+      Обновляет хранилище данными из словаря.
+
+   .. py:method:: get_all(self) -> dict
+
+      Возвращает все данные из хранилища.
+
+   .. py:method:: get_by_key(self, key: str) -> Any
+
+      Возвращает значение по ключу или `None`, если ключ не найден.
+
+   .. py:method:: delete_by_key(self, key: str) -> None
+
+      Удаляет значение по ключу. Вызывает `KeyError`, если ключ не найден.
+
+   .. py:method:: clear_all(self) -> None
+
+      Полностью очищает хранилище.

docs/root/api/index.rst

@@ -95,4 +95,5 @@
     router
     orchestrator/index
     command/index
-    response
+    response
+    bridge

docs/root/api/response.rst

@@ -5,7 +5,6 @@ Response
 
 `Response` — это объект, который передаётся в обработчик команды. Он создаётся автоматически при обработке пользовательского ввода и содержит статус валидации, введённые флаги, а также предоставляет механизм для обмена данными между обработчиками.
 
-`Response` наследует от `DataBridge` методы для работы с глобальным хранилищем, что позволяет обмениваться данными между обработчиками в рамках одной сессии.
 
 .. seealso::
 
@@ -57,99 +56,6 @@ Response
 
 -----
 
-Методы DataBridge
-
-`Response` наследует от `DataBridge` методы для работы с глобальным хранилищем, которое позволяет обмениваться данными между обработчиками в рамках одной сессии.
-
-update_data
-~~~~~~~~~~~
-
-.. code-block:: python
-   :linenos:
-
-   @classmethod
-   update_data(cls, data: dict[str, Any]) -> None
-
-Обновляет глобальное хранилище, добавляя или изменяя значения из переданного словаря.
-
-:param data: Словарь с данными для обновления хранилища
-:return: None
-
-Метод объединяет переданный словарь с данными в хранилище. Если ключ уже существует, его значение обновляется.
-
-**Пример использования:**
-
-.. literalinclude:: ../../code_snippets/response/snippet2.py
-   :linenos:
-   :language: python
-
------
-
-get_data
-~~~~~~~~
-
-.. code-block:: python
-   :linenos:
-
-   @classmethod
-   get_data(cls) -> dict[str, Any]
-
-Возвращает все данные из глобального хранилища.
-
-:return: Словарь со всеми данными из хранилища
-
-**Пример использования:**
-
-.. literalinclude:: ../../code_snippets/response/snippet3.py
-   :linenos:
-   :language: python
-
------
-
-clear_data
-~~~~~~~~~~
-
-.. code-block:: python
-   :linenos:
-
-   @classmethod
-   clear_data(cls) -> None
-
-Очищает глобальное хранилище.
-
-:return: None
-
-**Пример использования:**
-
-.. literalinclude:: ../../code_snippets/response/snippet4.py
-   :linenos:
-   :language: python
-
------
-
-delete_from_data
-~~~~~~~~~~~~~~~~
-
-.. code-block:: python
-   :linenos:
-
-   @classmethod
-   delete_from_data(cls, key: str) -> None
-
-Удаляет ключ и его значение из глобального хранилища.
-
-:param key: Ключ, который необходимо удалить из хранилища
-:return: None
-:raises KeyError: Если ключ не найден в хранилище.
-
-**Пример использования:**
-
-.. literalinclude:: ../../code_snippets/response/snippet5.py
-   :linenos:
-   :language: python
-
------
-
 Работа с флагами
 ----------------
 

src/argenta/data_bridge/__init__.py

@@ -0,0 +1,3 @@
+__all__ = ["DataBridge"]
+
+from .entity import DataBridge as DataBridge

src/argenta/data_bridge/entity.py

@@ -0,0 +1,23 @@
+__all__ = ["DataBridge"]
+
+from typing import Any
+
+
+class DataBridge:
+    def __init__(self, initial_data: dict[str, Any] | None = None) -> None:
+        self._data: dict[str, Any] = initial_data if initial_data else {}
+
+    def update(self, data: dict[str, Any]) -> None:
+        self._data.update(data)
+
+    def get_all(self) -> dict[str, Any]:
+        return self._data
+
+    def clear_all(self) -> None:
+        self._data.clear()
+
+    def get_by_key(self, key: str) -> Any:
+        return self._data.get(key)
+
+    def delete_by_key(self, key: str) -> None:
+        self._data.pop(key)

src/argenta/di/providers.py

@@ -4,6 +4,7 @@ __all__ = [
 
 from dishka import Provider, Scope, provide
 
+from argenta.data_bridge import DataBridge
 from argenta.orchestrator.argparser import ArgParser
 from argenta.orchestrator.argparser.entity import ArgSpace
 
@@ -16,3 +17,7 @@ class SystemProvider(Provider):
     @provide(scope=Scope.APP)
     def get_argspace(self) -> ArgSpace:
         return self._arg_parser.parsed_argspace
+
+    @provide(scope=Scope.APP)
+    def get_data_bridge(self) -> DataBridge:
+        return DataBridge()

src/argenta/response/entity.py

@@ -1,5 +1,4 @@
 __all__ = ["Response"]
-from typing import Any
 
 from dishka import Container
 
@@ -9,27 +8,7 @@ from argenta.response.status import ResponseStatus
 EMPTY_INPUT_FLAGS: InputFlags = InputFlags()
 
 
-class DataBridge:
-	_data: dict[str, Any] = {}
-
-	@classmethod
-	def update_data(cls, data: dict[str, Any]) -> None:
-		cls._data.update(data)
-
-	@classmethod
-	def get_data(cls) -> dict[str, Any]:
-		return cls._data
-
-	@classmethod
-	def clear_data(cls) -> None:
-		cls._data.clear()
-
-	@classmethod
-	def delete_from_data(cls, key: str) -> None:
-		cls._data.pop(key)
-
-
-class Response(DataBridge):
+class Response:
     _dishka_container: Container
 
     def __init__(

tests/unit_tests/test_response.py

@@ -1,131 +1,80 @@
 import unittest
 from datetime import date, datetime
 
-from argenta.command.flag import InputFlag
+from argenta.data_bridge import DataBridge
+from argenta.command.flag.models import InputFlag
 from argenta.command.flag.flags.models import InputFlags
-from argenta.response.entity import EMPTY_INPUT_FLAGS, DataBridge, Response
+from argenta.response.entity import EMPTY_INPUT_FLAGS, Response
 from argenta.response.status import ResponseStatus
 
 
 class TestDataBridge(unittest.TestCase):
     def setUp(self):
-        """Clear data before each test"""
-        DataBridge.clear_data()
-
-    def tearDown(self):
-        """Clear data after each test"""
-        DataBridge.clear_data()
+        """Create a new DataBridge instance for each test"""
+        self.data_bridge = DataBridge()
 
     def test_update_data_basic(self):
         """Test basic data update functionality"""
         test_data = {"key1": "value1", "key2": "value2"}
-        DataBridge.update_data(test_data)
-        self.assertEqual(DataBridge.get_data(), test_data)
+        self.data_bridge.update(test_data)
+        self.assertEqual(self.data_bridge.get_all(), test_data)
 
     def test_update_data_with_datetime(self):
         """Test updating data with datetime objects"""
         test_datetime = datetime(2024, 1, 15, 10, 30, 45)
         test_data = {"created_at": test_datetime, "name": "test"}
-        DataBridge.update_data(test_data)
-        
-        result = DataBridge.get_data()
+        self.data_bridge.update(test_data)
+
+        result = self.data_bridge.get_all()
         self.assertEqual(result["created_at"], test_datetime)
         self.assertEqual(result["name"], "test")
 
-    def test_update_data_with_date(self):
-        """Test updating data with date objects"""
-        test_date = date(2024, 1, 15)
-        test_data = {"birth_date": test_date, "active": True}
-        DataBridge.update_data(test_data)
-        
-        result = DataBridge.get_data()
-        self.assertEqual(result["birth_date"], test_date)
-        self.assertEqual(result["active"], True)
-
     def test_update_data_multiple_calls(self):
-        """Test multiple update_data calls merge data"""
-        first_data = {"key1": "value1", "date1": date(2024, 1, 1)}
-        second_data = {"key2": "value2", "date2": datetime(2024, 2, 1, 12, 0)}
-        
-        DataBridge.update_data(first_data)
-        DataBridge.update_data(second_data)
-        
-        result = DataBridge.get_data()
-        self.assertEqual(len(result), 4)
-        self.assertEqual(result["key1"], "value1")
-        self.assertEqual(result["key2"], "value2")
-        self.assertEqual(result["date1"], date(2024, 1, 1))
-        self.assertEqual(result["date2"], datetime(2024, 2, 1, 12, 0))
-
-    def test_update_data_overwrites_existing_keys(self):
-        """Test that update_data overwrites existing keys"""
-        initial_data = {"key": "old_value", "date": date(2024, 1, 1)}
-        updated_data = {"key": "new_value", "date": date(2024, 2, 1)}
-        
-        DataBridge.update_data(initial_data)
-        DataBridge.update_data(updated_data)
-        
-        result = DataBridge.get_data()
-        self.assertEqual(result["key"], "new_value")
-        self.assertEqual(result["date"], date(2024, 2, 1))
+        """Test multiple update calls merge data"""
+        first_data = {"key1": "value1"}
+        second_data = {"key2": "value2"}
+        self.data_bridge.update(first_data)
+        self.data_bridge.update(second_data)
+        self.assertEqual(len(self.data_bridge.get_all()), 2)
 
     def test_get_data_empty(self):
-        """Test get_data returns empty dict when no data"""
-        result = DataBridge.get_data()
-        self.assertEqual(result, {})
+        """Test get_all returns empty dict when no data"""
+        self.assertEqual(self.data_bridge.get_all(), {})
 
     def test_clear_data(self):
-        """Test clear_data removes all data"""
-        test_data = {"key": "value", "timestamp": datetime.now()}
-        DataBridge.update_data(test_data)
-        
-        # Verify data exists
-        self.assertNotEqual(DataBridge.get_data(), {})
-        
-        # Clear and verify
-        DataBridge.clear_data()
-        self.assertEqual(DataBridge.get_data(), {})
+        """Test clear_all removes all data"""
+        self.data_bridge.update({"key": "value"})
+        self.assertNotEqual(self.data_bridge.get_all(), {})
+        self.data_bridge.clear_all()
+        self.assertEqual(self.data_bridge.get_all(), {})
 
     def test_delete_from_data(self):
-        """Test delete_from_data removes specific key"""
-        test_data = {
-            "key1": "value1", 
-            "key2": "value2",
-            "created_at": datetime(2024, 1, 1, 10, 0)
-        }
-        DataBridge.update_data(test_data)
-        
-        # Delete one key
-        DataBridge.delete_from_data("key1")
-        
-        result = DataBridge.get_data()
-        self.assertEqual(len(result), 2)
+        """Test delete_by_key removes specific key"""
+        test_data = {"key1": "value1", "key2": "value2"}
+        self.data_bridge.update(test_data)
+        self.data_bridge.delete_by_key("key1")
+        result = self.data_bridge.get_all()
         self.assertNotIn("key1", result)
         self.assertIn("key2", result)
-        self.assertIn("created_at", result)
 
     def test_delete_from_data_nonexistent_key(self):
-        """Test delete_from_data with nonexistent key raises KeyError"""
-        test_data = {"existing_key": "value"}
-        DataBridge.update_data(test_data)
-        
+        """Test delete_by_key with nonexistent key raises KeyError"""
         with self.assertRaises(KeyError):
-            DataBridge.delete_from_data("nonexistent_key")
+            self.data_bridge.delete_by_key("nonexistent_key")
+
+    def test_get_by_key(self):
+        """Test get_by_key retrieves correct value"""
+        test_data = {"key1": "value1", "key2": date(2024, 1, 1)}
+        self.data_bridge.update(test_data)
+        self.assertEqual(self.data_bridge.get_by_key("key1"), "value1")
+        self.assertEqual(self.data_bridge.get_by_key("key2"), date(2024, 1, 1))
+        self.assertIsNone(self.data_bridge.get_by_key("nonexistent"))
 
 
 class TestResponse(unittest.TestCase):
-    def setUp(self):
-        """Clear data before each test"""
-        DataBridge.clear_data()
-
-    def tearDown(self):
-        """Clear data after each test"""
-        DataBridge.clear_data()
-
     def test_response_initialization_basic(self):
         """Test basic Response initialization"""
         response = Response(ResponseStatus.ALL_FLAGS_VALID)
-        
         self.assertEqual(response.status, ResponseStatus.ALL_FLAGS_VALID)
         self.assertEqual(response.input_flags, EMPTY_INPUT_FLAGS)
 
@@ -133,151 +82,9 @@ class TestResponse(unittest.TestCase):
         """Test Response initialization with input flags"""
         input_flags = InputFlags([InputFlag('test', input_value='value', status=None)])
         response = Response(ResponseStatus.INVALID_VALUE_FLAGS, input_flags)
-        
         self.assertEqual(response.status, ResponseStatus.INVALID_VALUE_FLAGS)
         self.assertEqual(response.input_flags, input_flags)
 
-    def test_response_inherits_databridge_functionality(self):
-        """Test that Response inherits DataBridge methods"""
-        response = Response(ResponseStatus.ALL_FLAGS_VALID)
-        test_data = {"message": "hello", "timestamp": datetime.now()}
-        
-        # Test update_data
-        response.update_data(test_data)
-        result = response.get_data()
-        self.assertEqual(result["message"], "hello")
-        self.assertIsInstance(result["timestamp"], datetime)
-
-    def test_response_data_passing_with_dates(self):
-        """Test passing date and datetime objects through Response"""
-        response = Response(ResponseStatus.ALL_FLAGS_VALID)
-        
-        current_time = datetime.now()
-        today = date.today()
-        
-        date_data = {
-            "current_datetime": current_time,
-            "current_date": today,
-            "custom_datetime": datetime(2024, 3, 15, 14, 30, 0),
-            "custom_date": date(2023, 12, 25),
-            "metadata": {"created": current_time, "updated": today}
-        }
-        
-        response.update_data(date_data)
-        retrieved_data = response.get_data()
-        
-        # Verify datetime objects are preserved
-        self.assertEqual(retrieved_data["current_datetime"], current_time)
-        self.assertEqual(retrieved_data["current_date"], today)
-        self.assertEqual(retrieved_data["custom_datetime"], datetime(2024, 3, 15, 14, 30, 0))
-        self.assertEqual(retrieved_data["custom_date"], date(2023, 12, 25))
-        
-        # Verify nested datetime objects
-        self.assertEqual(retrieved_data["metadata"]["created"], current_time)
-        self.assertEqual(retrieved_data["metadata"]["updated"], today)
-
-    def test_response_data_persistence_across_instances(self):
-        """Test that data persists across different Response instances"""
-        # First response instance
-        response1 = Response(ResponseStatus.ALL_FLAGS_VALID)
-        test_datetime = datetime(2024, 1, 1, 12, 0, 0)
-        response1.update_data({"session_start": test_datetime})
-        
-        # Second response instance
-        response2 = Response(ResponseStatus.UNDEFINED_FLAGS)
-        retrieved_data = response2.get_data()
-        
-        # Data should persist
-        self.assertEqual(retrieved_data["session_start"], test_datetime)
-
-    def test_response_data_complex_date_scenarios(self):
-        """Test complex scenarios with date/datetime handling"""
-        response = Response(ResponseStatus.ALL_FLAGS_VALID)
-        
-        # Create complex data structure with various date formats
-        complex_data = {
-            "user": {
-                "name": "John Doe",
-                "birth_date": date(1990, 5, 15),
-                "last_login": datetime(2024, 1, 15, 10, 30, 45),
-                "preferences": {
-                    "timezone": "UTC",
-                    "date_format": "%Y-%m-%d",
-                    "created_at": datetime(2023, 1, 1, 0, 0, 0)
-                }
-            },
-            "events": [
-                {"name": "login", "timestamp": datetime(2024, 1, 15, 10, 30, 45)},
-                {"name": "logout", "timestamp": datetime(2024, 1, 15, 18, 45, 30)},
-            ],
-            "dates_list": [
-                date(2024, 1, 1),
-                date(2024, 1, 2),
-                date(2024, 1, 3)
-            ]
-        }
-        
-        response.update_data(complex_data)
-        retrieved_data = response.get_data()
-        
-        # Verify all date/datetime objects are correctly preserved
-        self.assertEqual(retrieved_data["user"]["birth_date"], date(1990, 5, 15))
-        self.assertEqual(retrieved_data["user"]["last_login"], datetime(2024, 1, 15, 10, 30, 45))
-        self.assertEqual(retrieved_data["user"]["preferences"]["created_at"], datetime(2023, 1, 1, 0, 0, 0))
-        
-        # Verify dates in lists
-        self.assertEqual(len(retrieved_data["events"]), 2)
-        self.assertEqual(retrieved_data["events"][0]["timestamp"], datetime(2024, 1, 15, 10, 30, 45))
-        self.assertEqual(retrieved_data["events"][1]["timestamp"], datetime(2024, 1, 15, 18, 45, 30))
-        
-        # Verify date list
-        self.assertEqual(len(retrieved_data["dates_list"]), 3)
-        self.assertEqual(retrieved_data["dates_list"][0], date(2024, 1, 1))
-        self.assertEqual(retrieved_data["dates_list"][1], date(2024, 1, 2))
-        self.assertEqual(retrieved_data["dates_list"][2], date(2024, 1, 3))
-
-    def test_response_clear_data_functionality(self):
-        """Test clearing data functionality through Response"""
-        response = Response(ResponseStatus.ALL_FLAGS_VALID)
-        
-        # Add some data with dates
-        response.update_data({
-            "timestamp": datetime.now(),
-            "date": date.today(),
-            "message": "test"
-        })
-        
-        # Verify data exists
-        self.assertNotEqual(response.get_data(), {})
-        
-        # Clear data
-        response.clear_data()
-        
-        # Verify data is cleared
-        self.assertEqual(response.get_data(), {})
-
-    def test_response_delete_specific_date_data(self):
-        """Test deleting specific date-related data"""
-        response = Response(ResponseStatus.ALL_FLAGS_VALID)
-        
-        # Add mixed data
-        test_data = {
-            "start_date": date(2024, 1, 1),
-            "end_date": date(2024, 12, 31),
-            "created_at": datetime.now(),
-            "name": "test_session"
-        }
-        response.update_data(test_data)
-        
-        # Delete specific date field
-        response.delete_from_data("start_date")
-        
-        result = response.get_data()
-        self.assertNotIn("start_date", result)
-        self.assertIn("end_date", result)
-        self.assertIn("created_at", result)
-        self.assertIn("name", result)
-
     def test_response_status_types(self):
         """Test Response with different status types"""
         statuses = [
@@ -286,14 +93,10 @@ class TestResponse(unittest.TestCase):
             ResponseStatus.INVALID_VALUE_FLAGS,
             ResponseStatus.UNDEFINED_AND_INVALID_FLAGS
         ]
-        
         for status in statuses:
-            response = Response(status)
-            response.update_data({"timestamp": datetime.now(), "status_test": True})
-            
-            self.assertEqual(response.status, status)
-            self.assertIn("timestamp", response.get_data())
-            self.assertIn("status_test", response.get_data())
+            with self.subTest(status=status):
+                response = Response(status)
+                self.assertEqual(response.status, status)
 
 
 if __name__ == '__main__':