paso 01 — La Action y el pipeline

¡Hola, mundo!

Empecemos con una Action que no hace nada útil: imprime una línea y devuelve un stub vacío. Su valor está en otra parte: muestra el mínimo de declaraciones que AOA exige antes de ejecutar absolutamente nada.

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

▶ Probar en Colab ↗Abrir en GitHub ↗

salida

Son bastantes líneas para un simple saludo, pero ninguna es un ritual: cada una declara algo sin lo cual una Action en AOA no se considera completa. Todo empieza con un domain. @meta es el pasaporte de la Action; @check_roles(GuestRole) declara el acceso de forma explícita, nunca por defecto. Y @summary_aspect es el único punto de salida: sáltate cualquiera de los tres y la máquina se niega a ejecutar la Action, incluso antes de la primera llamada.

Params, Result y box

Los stubs están bien para las introducciones; una Action real trabaja con datos. Démosle una entrada y una salida, y cambiemos print por el instrumento que usan las Actions en AOA para hablar con el mundo exterior.

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

▶ Probar en Colab ↗Abrir en GitHub ↗

salida

box es más interesante que print: es un logger estructurado ligado al paso actual, que lleva el canal, el nivel, el domain, la action y el nombre del aspecto, por lo que se presta al filtrado, el enrutamiento y la exportación. Adónde va el evento no lo decide la Action: es cosa de la máquina, conectada desde fuera.

Múltiples aspectos

Un solo paso rara vez es suficiente. AOA no deja libertad para que los datos intermedios se escondan en variables locales o campos de objetos: fluyen a través del pipeline en un state explícito, mientras que la propia Action permanece vacía entre llamadas.

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

▶ Probar en Colab ↗Abrir en GitHub ↗

salida

El diccionario devuelto reemplaza state por completo — no lo extiende. @result_string("cleaned", required=True) es un verificador: un contrato verificable sobre la salida de un aspecto. Si se rompe la promesa, la máquina detiene el aspecto en el acto, sin dejar pasar nunca un state corrupto.

Herencia

Las Actions heredan como las clases normales, con una excepción deliberada: los aspectos no se heredan en el pipeline. La máquina construye el pipeline únicamente a partir de lo declarado en la propia clase.

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

▶ Probar en Colab ↗Abrir en GitHub ↗

salida

ChildOrderAction se ejecuta sin problemas, pero su pipeline solo contiene child_summary: el validate_aspect del padre nunca se ejecuta. ExtendedOrderAction lo hace bien: vuelve a declarar el aspecto y llama a super() para construir sobre la lógica del ancestro.

Experimentos

Este capítulo ha acumulado un buen número de reglas, pero todas comparten una propiedad: casi todas se verifican al declarar la clase, al importar el módulo, no al llamar a la Action. El código aquí es una especificación ejecutable; una violación del contrato sale a la luz antes de la primera ejecución, no una noche cualquiera en producción. Pruébalo: elige una forma de romper SayHelloAction del primer ejemplo y pulsa Ejecutar.

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

▶ Probar en Colab ↗Abrir en GitHub ↗

salida

Resumen

Ya tenemos el núcleo del modelo: una operación es una Action con un límite tipado (Params y Result), un pipeline lineal de aspectos y contratos que se verifican, en su mayoría, en la inicialización. El comportamiento está en la estructura del código, no en la documentación que lo describe por fuera.

Preguntas de repaso

¿Qué invariante garantiza la legibilidad «de una sola clase» de una Action, y en qué momento se verifica?
¿Por qué la ausencia de @check_roles es un error y no un silencioso «abierto a todos»?
El pipeline de AOA es lineal: sin ramificaciones, sin salidas laterales. ¿Qué gana esta restricción, y a qué costo?
¿Por qué los aspectos no se heredan automáticamente en el pipeline? Compáralo con la herencia normal de métodos en la POO.
¿Qué significa que «el código es una especificación ejecutable», y por qué la mayoría de los contratos se verifican en la inicialización y no en tiempo de ejecución?

← Paso 00 — Primeros pasos Paso 02 — State: la radiografía de la operación →