From f0a18e89c8e36ebbf61102f0eee5377a385e0ef1 Mon Sep 17 00:00:00 2001 From: kolo Date: Sun, 2 Nov 2025 18:34:33 +0300 Subject: [PATCH] docs --- docs/code_snippets/response/data_sharing.py | 49 ++++ docs/root/api/bridge.rst | 60 +++++ docs/root/api/index.rst | 3 +- docs/root/api/response.rst | 94 ------- src/argenta/data_bridge/__init__.py | 3 + src/argenta/data_bridge/entity.py | 23 ++ src/argenta/di/providers.py | 5 + src/argenta/response/entity.py | 23 +- tests/unit_tests/test_response.py | 279 +++----------------- 9 files changed, 184 insertions(+), 355 deletions(-) create mode 100644 docs/code_snippets/response/data_sharing.py create mode 100644 docs/root/api/bridge.rst create mode 100644 src/argenta/data_bridge/__init__.py create mode 100644 src/argenta/data_bridge/entity.py diff --git a/docs/code_snippets/response/data_sharing.py b/docs/code_snippets/response/data_sharing.py new file mode 100644 index 0000000..5f6c846 --- /dev/null +++ b/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("Вы и так не были аутентифицированы.") + diff --git a/docs/root/api/bridge.rst b/docs/root/api/bridge.rst new file mode 100644 index 0000000..6970c3c --- /dev/null +++ b/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 + + Полностью очищает хранилище. diff --git a/docs/root/api/index.rst b/docs/root/api/index.rst index e8be5b1..b53bc95 100644 --- a/docs/root/api/index.rst +++ b/docs/root/api/index.rst @@ -95,4 +95,5 @@ router orchestrator/index command/index - response \ No newline at end of file + response + bridge \ No newline at end of file diff --git a/docs/root/api/response.rst b/docs/root/api/response.rst index 2b3488b..9053db2 100644 --- a/docs/root/api/response.rst +++ b/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 - ------ - Работа с флагами ---------------- diff --git a/src/argenta/data_bridge/__init__.py b/src/argenta/data_bridge/__init__.py new file mode 100644 index 0000000..95adfdc --- /dev/null +++ b/src/argenta/data_bridge/__init__.py @@ -0,0 +1,3 @@ +__all__ = ["DataBridge"] + +from .entity import DataBridge as DataBridge diff --git a/src/argenta/data_bridge/entity.py b/src/argenta/data_bridge/entity.py new file mode 100644 index 0000000..6f1f3f6 --- /dev/null +++ b/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) diff --git a/src/argenta/di/providers.py b/src/argenta/di/providers.py index 8cd4d47..af6f379 100644 --- a/src/argenta/di/providers.py +++ b/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() diff --git a/src/argenta/response/entity.py b/src/argenta/response/entity.py index 9a902ac..516e80d 100644 --- a/src/argenta/response/entity.py +++ b/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__( diff --git a/tests/unit_tests/test_response.py b/tests/unit_tests/test_response.py index bc5cfa3..6a1d020 100644 --- a/tests/unit_tests/test_response.py +++ b/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__':