Orchestrator¶

Orchestrator — это высокоуровневый компонент, который конфигурирует и оркестрирует приложение, парсер командной строки, DI и остальные компоненты, находящиеся по иерархии на уровне с App.

В то время как App отвечает за логику интерактивной сессии (ввод команд, маршрутизация), Orchestrator подготавливает окружение для его работы и служит точкой входа в приложение.

Инициализация¶

DEFAULT_ARGPARSER: ArgParser = ArgParser(processed_args=[])

def __init__(self, arg_parser: ArgParser = DEFAULT_ARGPARSER,
             custom_providers: list[Provider] = [],
             auto_inject_handlers: bool = True) -> None

Создаёт и конфигурирует экземпляр Orchestrator.

arg_parser: Экземпляр ArgParser, отвечающий за парсинг аргументов командной строки при запуске скрипта (не путать с командами в интерактивном режиме).
custom_providers: Список пользовательских провайдеров dishka.Provider для добавления ваших сервисов (например, подключений к БД или API-клиентов) в di-контейнер.
auto_inject_handlers: Если True (по умолчанию), dishka автоматически внедрит зависимости в обработчики команд, инспектируя их сигнатуры.

Основные методы¶

start_polling(self, app: App) → None¶

Это главный метод, который запускает приложение. Он запускает бесконечный цикл ввода -> вывода.

Параметры:: app – Экземпляр App, который будет запущен.

Назначение и использование¶

Orchestrator абстрагирует сложность, связанную с настройкой DI и парсингом стартовых аргументов.

Такой подход разделяет ответственности: App отвечает за логику интерактивной сессии, а Orchestrator — за подготовку окружения и запуск приложения.

Пример использования:

import sqlite3
from sqlite3 import Connection
from typing import Iterable

from dishka import Provider, Scope, provide

from argenta import App, Orchestrator


class ConnectionProvider(Provider):
    @provide(scope=Scope.REQUEST)
    def new_connection(self) -> Iterable[Connection]:
        conn = sqlite3.connect(":memory:")
        yield conn
        conn.close()


# 2. Create and configure App
app = App()
# ... you can add routers here ...

# 3. Create Orchestrator, passing our provider
orchestrator = Orchestrator(custom_providers=[ConnectionProvider()])

# 4. Start the application
if __name__ == "__main__":
    orchestrator.start_polling(app)