Function read

python/pathway/io/python/__init__.py:510–618

Reads a table from a ConnectorSubject. Args: subject: An instance of a :py:class:`~pathway.python.ConnectorSubject`. schema: Schema of the resulting table. format: Deprecated. Pass values of proper types to ``subject``'s ``next`` instead. Format of the data p

(
    subject: ConnectorSubject,
    *,
    schema: type[Schema] | None = None,
    format: Literal["json", "raw", "binary"] | None = None,
    autocommit_duration_ms: int | None = 1500,
    debug_data=None,
    name: str | None = None,
    max_backlog_size: int | None = None,
    _stacklevel: int = 1,
    **kwargs,
)

@check_arg_types
@trace_user_frame
def read(
    subject: ConnectorSubject,
    *,
    schema: type[Schema] | None = None,
    format: Literal["json", "raw", "binary"] | None = None,
    autocommit_duration_ms: int | None = 1500,
    debug_data=None,
    name: str | None = None,
    max_backlog_size: int | None = None,
    _stacklevel: int = 1,
    **kwargs,
) -> Table:
    """Reads a table from a ConnectorSubject.

    Args:
        subject: An instance of a :py:class:`~pathway.python.ConnectorSubject`.
        schema: Schema of the resulting table.
        format: Deprecated. Pass values of proper types to ``subject``'s ``next`` instead.
            Format of the data produced by a subject, "json", "raw" or "binary". In case of
            a "raw"/"binary" format, table with single "data" column will be produced.
        debug_data: Static data replacing original one when debug mode is active.
        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
        name: A unique name for the connector. If provided, this name will be used in
            logs and monitoring dashboards. Additionally, if persistence is enabled, it
            will be used as the name for the snapshot that stores the connector's progress.
        max_backlog_size: Limit on the number of entries read from the input source and kept
            in processing at any moment. Reading pauses when the limit is reached and resumes
            as processing of some entries completes. Useful with large sources that
            emit an initial burst of data to avoid memory spikes. **Note**: The ``next``,
            ``next_json``, ``next_str``, and ``next_bytes`` methods of ``subject`` will
            block when the internal queue holding events before they are sent to the
            processing reaches ``max_backlog_size``. These methods will resume only when
            the queue size drops below this limit.

    Returns:
        Table: The table read.

    Example:

    >>> import pathway as pw
    >>> from pathway.io.python import ConnectorSubject
    >>>
    >>> class MySchema(pw.Schema):
    ...     a: int
    ...     b: str
    ...
    >>>
    >>> class MySubject(ConnectorSubject):
    ...     def run(self) -> None:
    ...         for i in range(4):
    ...             self.next(a=i, b=f"x{i}")
    ...     @property
    ...     def _deletions_enabled(self) -> bool:
    ...         return False
    ...
    >>>