الخطوة 01 / 28

Action والـ pipeline

يدور كل شيء في AOA حول كيان واحد: Action. إنها عملية أعمال مصوغة على هيئة فئة: مدخل واحد مُنمَّط (Params)، ومخرج واحد (Result)، وبينهما سلسلة خطوات مباشرة تُقرأ من الأعلى إلى الأسفل، كما تُقرأ صفحة. يقابل كل قسم أدناه ملفًا حقيقيًا في examples/step_01_Action_and_pipeline/ — اضغط تشغيل لترى ناتجه الفعلي المُلتقط، أو افتحه في Colab وشغّله بنفسك.

مرحبًا بالعالم!

لنبدأ بـ Action لا يفعل شيئًا مفيدًا — يطبع سطرًا ويُعيد stub فارغًا. قيمته في مكان آخر: فهو يُظهر الحد الأدنى من الإعلانات التي بدونها لن يُشغّل 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
الناتج

هذا عدد لا بأس به من الأسطر لتحية بهذه البساطة — لكن لا واحد منها طقس شكلي: كل واحد يُعلن أمرًا لا يُعدّ الـ Action في AOA مكتملًا بدونه. يبدأ كل شيء بـdomain. @meta هو جواز سفر الـ Action؛ @check_roles(GuestRole) يُعلن الوصول صراحةً، لا افتراضيًا. أما @summary_aspect فهو نقطة الخروج الوحيدة: تجاوز أيًّا من الثلاثة، وسترفض الآلة تشغيل الـ Action حتى قبل أول استدعاء.

Params، Result، وbox

الـ stubs جيدة للمقدمات، أما الـ 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
الناتج

box أكثر إثارة للاهتمام من print — فهو logger مُهيكَل مرتبط بالخطوة الحالية: يحمل قناة، ومستوى، وdomain، واسم action، وaspect، لذا يسهل إخضاعه للتصفية والتوجيه والتصدير. أما إلى أين يذهب الحدث فذلك ليس قرار الـ Action نفسه: تلك مسؤولية الآلة، المُوصَّلة من الخارج.

عدة aspects

نادرًا ما تكفي خطوة واحدة. لا يترك AOA للبيانات الوسيطة حرية الاختباء في متغيرات محلية أو حقول كائن — بل تمر عبر الـ pipeline ضمن 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
الناتج

القاموس المُعاد يستبدل بالكامل state — لا يُضيف إليه. @result_string("cleaned", required=True) هو مُتحقق: عقد قابل للتحقق على ناتج aspect. أخلّ بالوعد، وستوقِف الآلة الـ aspect فورًا، دون أن تدع state تالفة تمرّ أبعد من ذلك.

الوراثة

تُورَّث الـ Actions مثل الفئات العادية — باستثناء واحد متعمَّد: الـ aspects لا تُورَّث في الـ pipeline. تبني الآلة الـ pipeline فقط مما هو مُعلَن في الفئة نفسها.

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
الناتج

يعمل ChildOrderAction بشكل ممتاز، لكن الـ pipeline الخاص به لا يحتوي إلا على child_summary — أما validate_aspect الخاص بالأصل فلا يُنفَّذ أبدًا. أما ExtendedOrderAction فيفعل كل شيء بشكل صحيح: فهو يُعلن الـ aspect من جديد ويستدعي super() للاستناد إلى منطق الأصل.


التجارب

تراكم في هذا الفصل عدد لا بأس به من القواعد — لكنها تشترك في خاصية واحدة: يُتحقَّق من معظمها عند إعلان الفئة، عند استيراد الوحدة، لا عند استدعاء الـ Action. الكود هنا مواصفة قابلة للتنفيذ؛ يظهر خرق العقد قبل أول تشغيل، لا في ليلة ما وسط بيئة الإنتاج. جرّب بنفسك — ابحث عن طريقة لكسر SayHelloAction من المثال الأول واضغط تشغيل.

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
الناتج

الخلاصة

لقد تشكّل العمود الفقري للنموذج بالفعل: العملية هي Action ذو حد مُنمَّط (Params وResult)، وpipeline خطي من الـ aspects، وعقود يُتحقَّق من معظمها عند التهيئة. يُعبَّر عن السلوك في بنية الكود، لا في توثيق قائم بمعزل عنه.


أسئلة المراجعة

  1. أي ثابت يجعل الـ Action قابلاً للقراءة «من فئة واحدة»، وفي أي لحظة يُتحقَّق من ذلك؟
  2. لماذا يُعدّ غياب @check_roles خطأً، لا «إتاحة صامتة للجميع»؟
  3. الـ pipeline في AOA خطي — دون تفرعات أو مخارج جانبية. ماذا يمنحنا هذا القيد، وبأي ثمن؟
  4. لماذا لا تُورَّث الـ aspects في الـ pipeline تلقائيًا؟ قارن ذلك بالوراثة المعتادة للدوال في البرمجة الشيئية.
  5. ماذا تعني عبارة «الكود مواصفة قابلة للتنفيذ»، ولماذا يُتحقَّق من معظم العقود عند التهيئة لا في وقت التشغيل؟
خطوة 01 — Action والـ pipeline · aoa.run