dora/examples/python-operator-dataflow/file_saver_op.py

import pyarrow as pa

from dora import DoraStatus


class Operator:
    """
    Infering object from images
    """

    def __init__(self):
        self.last_file = ""
        self.last_path = ""
        self.last_netadata = None

    def on_event(
        self,
        dora_event,
        send_output,
    ) -> DoraStatus:
        if dora_event["type"] == "INPUT" and dora_event["id"] == "file":
            input = dora_event["value"][0].as_py()

            with open(input["path"], "r") as file:
                self.last_file = file.read()
                self.last_path = input["path"]
                self.last_metadata = dora_event["metadata"]
            with open(input["path"], "w") as file:
                file.write(input["raw"])

            send_output(
                "saved_file",
                pa.array(
                    [
                        {
                            "raw": input["raw"],
                            "path": input["path"],
                            "origin": dora_event["id"],
                        }
                    ]
                ),
                dora_event["metadata"],
            )
        return DoraStatus.CONTINUE