Function _json_table_from_list

python/pathway/tests/test_json.py:27–46

(data)

def _json_table_from_list(data):
    class _JsonSubject(pw.io.python.ConnectorSubject):
        def __init__(self, data: list[dict[str, Any]]) -> None:
            super().__init__()
            self.data = data

        def run(self) -> None:
            for key, row in enumerate(self.data):
                self.next(
                    key=key + 1, **{name: pw.Json(value) for name, value in row.items()}
                )

    schema = pw.schema_builder(
        columns={
            "key": pw.column_definition(dtype=int, primary_key=True),
            **{name: pw.column_definition(dtype=pw.Json) for name in data[0]},
        }
    )

    return pw.io.python.read(_JsonSubject(data), schema=schema).without(pw.this.key)


def _json_table(**kwargs) -> pw.Table: