PaeoniaCommon/SkillsOdea

Criar um Processo

Todos os caminhos são relativos à raiz do repositório. O pacote está em process_package/.

Como a classe pai funciona

process_package/toy_process/process.py define Process:

• __init__(**overrides) define self.params = {**self.set_parameters(), **overrides}, então self.steps = self.set_steps().
• set_parameters() e set_steps() são abstratos. Uma filha que não tenha um deles não pode ser instanciada.
• run(context=None) itera sobre self.steps em ordem:
  • uma instância de Process → step.run(context), aninhada, com o mesmo dicionário de contexto
  • qualquer outro chamável → step(context)
  • qualquer outra coisa → TypeError
• run retorna o dicionário de contexto e imprime um trace indentado.

Não modifique toy_process/ nem sobrescreva __init__ ou run em uma filha.

Regras para a classe filha

1. from toy_process import Process e crie uma subclasse.
2. set_parameters(self) retorna um dict de valores padrão. Use {} se não houver nenhum.
3. set_steps(self) retorna uma list. Cada item é:
   • um método vinculado: self.count_words, sem parênteses, ou
   • uma instância de Process: CleanText(lowercase=self.params["lowercase"]).
4. Cada método de etapa tem a assinatura def nome(self, context):.
   • Leia configurações de self.params["..."].
   • Leia e escreva dados em context["..."].
   • Não retorne nada. Valores retornados são ignorados.
5. Defina uma classe de subprocesso acima da classe que a utiliza, ou importe-a.

Template

"""Descrição de uma linha do que o processo faz.

Execute de process_package/:
    python -m examples.<nome_do_modulo>
"""

from toy_process import Process


class SubTask(Process):
    def set_parameters(self):
        return {"option": True}

    def set_steps(self):
        return [self.do_part]

    def do_part(self, context):
        if self.params["option"]:
            context["value"] = context["value"] * 2


class MainTask(Process):
    def set_parameters(self):
        return {"start": 1, "option": True}

    def set_steps(self):
        return [
            self.load,
            SubTask(option=self.params["option"]),  # subprocesso aninhado
            self.finish,
        ]

    def load(self, context):
        context["value"] = self.params["start"]

    def finish(self, context):
        context["result"] = f"value is {context['value']}"


if __name__ == "__main__":
    print(MainTask(start=5).run()["result"])

Arquivos a criar

1. process_package/examples/<nome_do_modulo>.py: o processo, a partir do template.
2. process_package/tests/test_<nome_do_modulo>.py: um teste unittest que chama .run() e faz asserções nas chaves do contexto retornado. Verifique também que sobrescrever um parâmetro altera o resultado.

import unittest

from examples.<nome_do_modulo> import MainTask


class MainTaskTests(unittest.TestCase):
    def test_run(self):
        self.assertEqual(MainTask(start=5).run()["result"], "value is 10")

    def test_parameter_override(self):
        self.assertEqual(MainTask(start=5, option=False).run()["result"], "value is 5")


if __name__ == "__main__":
    unittest.main()

Em seguida, verifique com a skill run-process. Um exemplo completo funcional é process_package/examples/word_report.py.