шаг 01 — Action и пайплайн

Hello, world!

Начнём с Action, который не делает ничего полезного — печатает строку и возвращает пустую заглушку. Его ценность в другом: он показывает минимальный набор объявлений, без которых AOA не запустит вообще ничего.

01_hello_world.py

class GreetingDomain(BaseDomain):    name = "greeting"    description = "Greetings domain"@meta(description="Say hello to the world", domain=GreetingDomain)@check_roles(GuestRole)class SayHelloAction(BaseAction[ParamsStub, ResultStub]):    @summary_aspect("Print greeting and return stub")    async def output_summary(self, params, state, box, connections):        print("Hello, world!")        return ResultStub()

$ uv run python examples/step_01_Action_and_pipeline/01_hello_world.py

▶ Попробовать в Colab ↗Открыть в GitHub ↗

вывод

Для такого простого приветствия строк набралось немало — но ни одна из них не ритуал: каждая объявляет нечто, без чего Action в AOA не считается завершённым. Всё начинается с domain. @meta — это паспорт Action; @check_roles(GuestRole) объявляет доступ вслух, а не по умолчанию. А @summary_aspect — единственная точка выхода: пропустите любое из трёх, и машина откажется запускать Action ещё до первого вызова.

Params, Result и box

Заглушки хороши для введения, а вот настоящий Action работает с данными. Добавим ему вход и выход — и заменим print на инструмент, которым Action в AOA обращается к внешнему миру.

02_params_result_and_box.py

class GreetParams(BaseParams):    name: str = Field(description="Name of the person to greet")class GreetResult(BaseResult):    message: str = Field(description="Assembled greeting message")@meta(description="Greet a person by name", domain=GreetingDomain)@check_roles(GuestRole)class GreetPersonAction(BaseAction[GreetParams, GreetResult]):    @summary_aspect("Build greeting and return result")    async def greet_summary(self, params, state, box, connections):        await box.info(            Channel.business,            "Greeting: Hello, {%var.name|cyan}!",            name=params.name,        )        return GreetResult(message=f"Hello, {params.name}!")

$ uv run python examples/step_01_Action_and_pipeline/02_params_result_and_box.py

▶ Попробовать в Colab ↗Открыть в GitHub ↗

вывод

box интереснее, чем print, — это структурированный логгер, привязанный к текущему шагу: он несёт канал, уровень, domain, имя action и aspect, поэтому легко поддаётся фильтрации, маршрутизации и экспорту. Куда уходит событие — не решение самого Action: это забота машины, подключаемая снаружи.

Несколько аспектов

Одного шага редко бывает достаточно. AOA не оставляет промежуточным данным свободы прятаться в локальных переменных или полях объекта — они проходят через пайплайн в явном state, тогда как сам Action остаётся пустым между вызовами.

03_multiple_aspects.py

@meta(description="Process input string through multiple steps", domain=ProcessingDomain)@check_roles(GuestRole)class ProcessInputAction(BaseAction[ProcessParams, ProcessResult]):    @regular_aspect("Step 1: Strip whitespace and lowercase")    @result_string("cleaned", required=True)    async def validate_aspect(self, params, state, box, connections):        cleaned = params.raw_input.strip().lower()        return {"cleaned": cleaned}    @regular_aspect("Step 2: Enrich data")    @result_string("cleaned", required=True)    @result_string("enriched", required=True)    async def enrich_aspect(self, params, state, box, connections):        enriched = f"enriched::{state['cleaned']}"        return {"cleaned": state["cleaned"], "enriched": enriched}    @summary_aspect("Step 3: Assemble final result")    async def assemble_summary(self, params, state, box, connections):        return ProcessResult(            cleaned=state["cleaned"],            enriched=state["enriched"],            final=f"{state['cleaned']} → {state['enriched']}",        )

$ uv run python examples/step_01_Action_and_pipeline/03_multiple_aspects.py

▶ Попробовать в Colab ↗Открыть в GitHub ↗

вывод

Возвращённый словарь полностью заменяет state — а не дополняет его. @result_string("cleaned", required=True) — это checker: проверяемый контракт на выход аспекта. Нарушьте обещание — и машина немедленно остановит аспект, не дав испорченному state пройти дальше.

Наследование

Action наследуются как обычные классы — с одним намеренным исключением: аспекты не наследуются в пайплайн. Машина строит пайплайн только из того, что объявлено в самом классе.

04_inheritance.py

# Parent: two aspects in the pipelineclass BaseOrderAction(BaseAction[OrderParams, OrderResult]):    @regular_aspect("Validate order")    async def validate_aspect(self, ...): ...    @summary_aspect("Base result")    async def base_summary(self, ...): ...# Child: declares only its own summary — validate_aspect will NOT runclass ChildOrderAction(BaseOrderAction):    @summary_aspect("Child result")    async def child_summary(self, ...): ...# The right way: declare the aspect explicitly and call super()class ExtendedOrderAction(BaseOrderAction):    @regular_aspect("Validate order")    @result_instance("steps", list, required=True)    async def validate_aspect(self, params, state, box, connections):        result = await super().validate_aspect(params, state, box, connections)        return {**result, "extended": True}    @summary_aspect("Extended result")    async def extended_summary(self, ...): ...

$ uv run python examples/step_01_Action_and_pipeline/04_inheritance.py

▶ Попробовать в Colab ↗Открыть в GitHub ↗

вывод

ChildOrderAction прекрасно работает, но его пайплайн содержит только child_summary — родительский validate_aspect никогда не выполняется. ExtendedOrderAction делает всё правильно: он заново объявляет аспект и вызывает super(), чтобы опереться на логику предка.

Эксперименты

В этой главе накопилось немало правил — но у них есть общее свойство: почти все они проверяются при объявлении класса, при импорте модуля, а не при вызове Action. Код здесь — исполняемая спецификация; нарушение контракта всплывает до первого запуска, а не однажды ночью в продакшене. Попробуйте сами — найдите способ сломать SayHelloAction из первого примера и нажмите Run.

01_hello_world.py

class GreetingDomain(BaseDomain):    name = "greeting"    description = "Greetings domain"@meta(description="Say hello to the world", domain=GreetingDomain)@check_roles(GuestRole)class SayHelloAction(BaseAction[ParamsStub, ResultStub]):    @summary_aspect("Print greeting and return stub")    async def output_summary(self, params, state, box, connections):        print("Hello, world!")        return ResultStub()

$ uv run python examples/step_01_Action_and_pipeline/01_hello_world.py

▶ Попробовать в Colab ↗Открыть в GitHub ↗

вывод

Итоги

Костяк модели уже сложился: операция — это Action с типизированной границей (Params и Result), линейным пайплайном аспектов и контрактами, которые в основном проверяются при инициализации. Поведение выражено в структуре кода, а не в документации, которая существует отдельно от него.

Контрольные вопросы

Какой инвариант делает Action читаемым «по одному классу», и в какой момент это проверяется?
Почему отсутствие @check_roles — это ошибка, а не молчаливое «доступно всем»?
Пайплайн AOA линеен — без ветвлений и побочных выходов. Что даёт это ограничение и какой ценой?
Почему аспекты не наследуются в пайплайн автоматически? Сравните с обычным наследованием методов в ООП.
Что значит «код — исполняемая спецификация», и почему большинство контрактов проверяется при инициализации, а не во время выполнения?

← Шаг 00 — Начало работы Шаг 02 — State: рентген операции →