0.4.0

What's Changed

• feat: cli dream upgrade by @craycrayfish in #151
• doc: update aineko stream command by @Barabazs in #157
• feat: implement abstract dataset and refactor Kafka dataset by @giddyphysicist in #150
• refactor: dataset implementation & documentation by @Barabazs in #161
• refactor: internal models by @Barabazs in #162
• feat: decouple KafkaDataset from core logic (Runner, AbstractNode, Config,...) by @Barabazs in #163

Breaking Changes

We've redesigned the dataset implementation to be more flexible, and decoupled it from the core logic. This will allow for easier integration of new datasets in the future.

Please note the following breaking changes:

• pipeline.yml: The type field of the dataset section should now be the dotted path of the dataset class, rather than the name of the dataset type. For example, aineko.datasets.kafka.KafkaDataset instead of kafka_stream.
• running nodes in test mode:

  • The setup_test method of the AbstractNode class has been updated to include a dataset_type parameter. This is the same value as the type field in the pipeline.yml file.

  • The output messages of the run_test now return the complete message object, rather than just the message value. This is to allow for easier testing of the message metadata, such as the timestamp and key. If you only want the message value, you can use the message_helper fixture to extract the value from the message object.

    Before and After code examples

    Before:

    def test_mynode():
        """Unit test for MySumNode."""
    
        mynode = MySumNode(
            node_name="MySumNode",
            pipeline_name="test_pipeline",
            test=True
        )
        mynode.setup_test(
            inputs={
                "test_sequence": [1, 2, 3]
            },  # input a list of elements to be read
            outputs=["test_sum"],  # list of dataset names that are written to
            params={"initial_state": 0, "increment": 1},
        )
        outputs = mynode.run_test()
        assert outputs["test_sum"] == [2, 3, 4]
        assert mynode.state == 4

    After:

    def test_mynode(message_helper):
        """Unit test for MySumNode."""
    
        mynode = MySumNode(
            node_name="MySumNode",
            pipeline_name="test_pipeline",
            test=True
        )
        mynode.setup_test(
            dataset_type="aineko.datasets.kafka.KafkaDataset", # The dataset type to use
            inputs={
                "test_sequence": [1, 2, 3]
            },  # input a list of elements to be read
            outputs=["test_sum"],  # list of dataset names that are written to
            params={"initial_state": 0, "increment": 1},
        )
        outputs = mynode.run_test()
        assert message_helper(outputs["test_sum"]) == [2, 3, 4]
        assert mynode.state == 4

Full Changelog: 0.3.2...0.4.0