Function pipeline

python/pathway/tests/test_async_transformer.py:216–254

()

    counter = mock.Mock()

    def pipeline():
        G.clear()

        class OutputSchema(pw.Schema):
            ret: int

        class TestAsyncTransformer(pw.AsyncTransformer, output_schema=OutputSchema):
            async def invoke(self, value: int) -> dict[str, Any]:
                counter()
                await asyncio.sleep(random.uniform(0, 0.1))
                return dict(ret=value + 1)

        input = T(
            """
                | value
            1   | 1
            2   | 2
            3   | 3
            """
        )
        expected = T(
            """
                | ret
            1   | 2
            2   | 3
            3   | 4
            """
        )

        result = TestAsyncTransformer(input_table=input).successful

        assert_table_equality(
            result,
            expected,
            persistence_config=pw.persistence.Config(
                pw.persistence.Backend.filesystem(cache_dir),
                persistence_mode=pw.PersistenceMode.SELECTIVE_PERSISTING,
            ),
        )

    # run twice to check if cache is used
    pipeline()