Function replay_csv

python/pathway/demo/__init__.py:212–254

Replay a static CSV files as a data stream. Args: path: Path to the file to stream. schema: Schema of the resulting table. autocommit_duration_ms: the maximum time between two commits. Every autocommit_duration_ms milliseconds, the updates received by the con

(
    path: str | PathLike,
    *,
    schema: type[pw.Schema],
    input_rate: float = 1.0,
)

def replay_csv(
    path: str | PathLike,
    *,
    schema: type[pw.Schema],
    input_rate: float = 1.0,
) -> pw.Table:
    """Replay a static CSV files as a data stream.

    Args:
        path: Path to the file to stream.
        schema: Schema of the resulting table.
        autocommit_duration_ms: the maximum time between two commits. Every
            autocommit_duration_ms milliseconds, the updates received by the connector are
            committed and pushed into Pathway Live Data Framework's computation graph.
        input_rate (float, optional): The rate at which rows are read per second. Defaults to 1.0.

    Returns:
        Table: The table read.

    Note: the CSV files should follow a standard CSV settings. The separator is ',', the
    quotechar is '"', and there is no escape.

    """

    autocommit_ms = int(1000.0 / input_rate)

    columns = set(schema.column_names())

    class FileStreamSubject(pw.io.python.ConnectorSubject):
        def run(self):
            with open(path, newline="") as csvfile:
                csvreader = csv.DictReader(csvfile)
                for row in csvreader:
                    values = {key: row[key] for key in columns}
                    self.next_json(values)
                    time.sleep(1.0 / input_rate)

    return pw.io.python.read(
        FileStreamSubject(datasource_name="demo.replay-csv"),
        schema=schema.with_types(**{name: str for name in schema.column_names()}),
        autocommit_duration_ms=autocommit_ms,
        format="json",
    ).cast_to_types(**schema.typehints())


def replay_csv_with_time(