pw.io.python

An abstract class allowing to create custom python connectors.

Custom python connector can be created by extending this class and implementing run() function responsible for filling the buffer with data. This function will be started by pathway engine in a separate thread.

In order to send a message next() method can be used.

If the subject won’t delete records, set the class property deletions_enabled to False as it may help to improve the performance.

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


s = MySubject()

table = pw.io.python.read(s, schema=MySchema)
pw.debug.compute_and_print(table, include_id=False)

a | b
0 | x0
1 | x1
2 | x2
3 | x3

close()

sourceSends a sentinel message.

Should be called to indicate that no new messages will be sent.

commit()

sourceSends a commit message.

end()

sourceJoins a thread running run().

Should not be called directly.

next(**kwargs)

sourceSends a message to the enigne.

The arguments should be compatible with the schema passed to read(). Values for all fields should be passed to this method unless they have a default value specified in the schema.

Example:

import pathway as pw
import pandas as pd

class InputSchema(pw.Schema):
    a: pw.DateTimeNaive
    b: bytes
    c: int

class InputSubject(pw.io.python.ConnectorSubject):
    def run(self):
        self.next(a=pd.Timestamp("2021-03-21T18:34:12"), b="abc".encode(), c=3)
        self.next(a=pd.Timestamp("2022-04-01T11:12:12"), b="def".encode(), c=42)

t = pw.io.python.read(InputSubject(), schema=InputSchema)
pw.debug.compute_and_print(t, include_id=False)

a                   | b      | c
2021-03-21 18:34:12 | b'abc' | 3
2022-04-01 11:12:12 | b'def' | 42

next_bytes(message)

sourceSends a message.

• Parameters
  message (bytes) – a message represented as bytes.

next_json(message)

sourceSends a message.

• Parameters
  message (dict) – Dict representing json.

next_str(message)

sourceSends a message.

• Parameters
  message (str) – a message represented as a string.

on_persisted_run()

sourceThis method is called by Rust core to notify that the state will be persisted in this run.

on_stop()

sourceCalled after the end of the run() function.

final seek(state)

sourceCalled by Rust core on start to resume reading from the last stopping point.

start()

sourceRuns a separate thread with function feeding data into buffer.

Should not be called directly.

close()

sourceSends a sentinel message.

Should be called to indicate that no new messages will be sent.

commit()

sourceSends a commit message.

end()

sourceJoins a thread running run().

Should not be called directly.

next(**kwargs)

sourceSends a message to the enigne.

The arguments should be compatible with the schema passed to read(). Values for all fields should be passed to this method unless they have a default value specified in the schema.

Example:

import pathway as pw
import pandas as pd

class InputSchema(pw.Schema):
    a: pw.DateTimeNaive
    b: bytes
    c: int

class InputSubject(pw.io.python.ConnectorSubject):
    def run(self):
        self.next(a=pd.Timestamp("2021-03-21T18:34:12"), b="abc".encode(), c=3)
        self.next(a=pd.Timestamp("2022-04-01T11:12:12"), b="def".encode(), c=42)

t = pw.io.python.read(InputSubject(), schema=InputSchema)
pw.debug.compute_and_print(t, include_id=False)

a                   | b      | c
2021-03-21 18:34:12 | b'abc' | 3
2022-04-01 11:12:12 | b'def' | 42

next_bytes(message)

sourceSends a message.

• Parameters
  message (bytes) – a message represented as bytes.

next_json(message)

sourceSends a message.

• Parameters
  message (dict) – Dict representing json.

next_str(message)

sourceSends a message.

• Parameters
  message (str) – a message represented as a string.

on_persisted_run()

sourceThis method is called by Rust core to notify that the state will be persisted in this run.

on_stop()

sourceCalled after the end of the run() function.

seek(state)

sourceCalled by Rust core on start to resume reading from the last stopping point.

start()

sourceRuns a separate thread with function feeding data into buffer.

Should not be called directly.