Function test_python_connector_upsert_json

python/pathway/tests/test_io.py:2494–2531

(tmp_path: pathlib.Path)

def test_python_connector_upsert_json(tmp_path: pathlib.Path):
    output_path = tmp_path / "output.csv"

    class TestSubject(pw.io.python.ConnectorSubject):
        @property
        def _session_type(self) -> SessionType:
            return SessionType.UPSERT

        def run(self):
            self._add(
                api.ref_scalar(0),
                json.dumps({"word": "one", "digit": 1}).encode("utf-8"),
            )
            time.sleep(5e-2)
            self._add(
                api.ref_scalar(0),
                json.dumps({"word": "two", "digit": 2}).encode("utf-8"),
            )
            time.sleep(5e-2)
            self._add(
                api.ref_scalar(0),
                json.dumps({"word": "three", "digit": 3}).encode("utf-8"),
            )
            self.close()

    class InputSchema(pw.Schema):
        word: str
        digit: int

    table = pw.io.python.read(
        TestSubject(), format="json", schema=InputSchema, autocommit_duration_ms=10
    )
    pw.io.csv.write(table, output_path)
    run()

    result = pd.read_csv(output_path)
    assert len(result) == 5
    assert_sets_equality_from_path(output_path, {"three,3,1"})


def test_python_connector_upsert_remove_json(tmp_path: pathlib.Path):