Crear un Proceso

¿Qué hace Crear un Proceso?

Todas las rutas son relativas a la raíz del repositorio. El paquete está en process_package/.

Cómo funciona la clase padre

process_package/toy_process/process.py define Process:

• __init__(**overrides) establece self.params = {**self.set_parameters(), **overrides}, luego self.steps = self.set_steps().
• set_parameters() y set_steps() son abstractos. Una clase hija que falte a uno de ellos no se puede instanciar.
• run(context=None) itera sobre self.steps en orden:
  • una instancia de Process → step.run(context), anidado, con el mismo diccionario context
  • cualquier otro callable → step(context)
  • cualquier otra cosa → TypeError
• run devuelve el diccionario context e imprime un trace indentado.

No modifiques toy_process/ ni sobrescribas __init__ o run en una clase hija.

Reglas para la clase hija

1. from toy_process import Process y crear subclase de ella.
2. set_parameters(self) devuelve un dict con valores predeterminados. Usa {} si no hay ninguno.
3. set_steps(self) devuelve una list. Cada elemento es:
   • un método enlazado: self.count_words, sin paréntesis, o
   • una instancia de Process: CleanText(lowercase=self.params["lowercase"]).
4. Cada método de paso tiene la firma def nombre(self, context):.
   • Lee configuraciones de self.params["..."].
   • Lee y escribe datos en context["..."].
   • No devuelve nada. Los valores de retorno se ignoran.
5. Define una clase de subproceso arriba de la clase que la usa, o impórtala.

Plantilla

"""Descripción de una línea de lo que hace el proceso.

Ejecutar desde process_package/:
    python -m examples.<nombre_módulo>
"""

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"]),  # subproceso anidado
            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"])

Archivos a crear

1. process_package/examples/<nombre_módulo>.py: el proceso, a partir de la plantilla.
2. process_package/tests/test_<nombre_módulo>.py: una prueba unittest que llama a .run() y verifica las claves en el context devuelto. También comprueba que sobrescribir un parámetro cambie el resultado.

import unittest

from examples.<nombre_módulo> 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()

Luego verifica con la habilidad run-process. Un ejemplo completo y funcional es process_package/examples/word_report.py.