## io.Source A table you read yourself: subclass it and implement [read](odcs.read.md#sverify.odcs.read). Usage ``` python io.Source( name, *, schema=None, assumes=None, on_violation=None, ) ``` `read(since)` returns what is new since the cursor of the last read (a [Batch](io.Batch.md#sverify.io.Batch)), or None if nothing is. The first read gets `since=None` and returns the whole table. Each batch that changes something is a new version of the source (1, 2, …), and a refresh hands the nodes reading it exactly the change between their version and the latest: - `kind="append"` batches are the rows added. That rests on an assumption, such as "rows are only ever appended, in order of `ts`": state it as `assumes`, and the refresh report repeats it. `refresh(audit=True)` reads the whole table again ([reread](io.Source.md#sverify.io.Source.reread)) and raises [AuditError](errors.AuditError.md#sverify.errors.AuditError) where the assumption broke. - `kind="snapshot"` batches are the whole table. The source diffs each against the last, so the change is exact and nothing is assumed. - `kind="changes"` batches are rows added and removed, trusted as read. The source keeps the table as of its latest version in memory, since joins and keys need it, and has no metadata: preconditions on it are checked on the rows. Versions count reads, and a sink records the version of each source next to its cursor. A restarted job reads the whole table again, unless the source implements [restore](io.Source.md#sverify.io.Source.restore): then it starts from the table as of the cursor its sink recorded, and the first refresh reads only what is new since (`read(cursor)`). A stored node reading a source that cannot restore is recomputed and written again in full, with versions above the ones its sink recorded, so its sink still writes exactly once. ## Parameters `name: str` The table's name in the plans that read it. `schema: pa.Schema | None = None` The table's schema. With it, building a node proves the transform's contract without reading the source, so the first read happens at the first refresh; it is needed when the first read may return nothing. Default: the first batch's. `assumes: str | None = None` What the batches rest on, such as `"rows are only appended, in order of ts"`. Default: None. `on_violation: Literal[``"error", ``"quarantine"``, ``"hold"``] | None`` = None` What a refresh does with a batch that breaks a precondition on this source, for transforms that set no `on_violation` of their own: `"error"`, `"quarantine"` or `"hold"` (see [sverify.transform](transform.md#sverify.transform)). Default: None, `"error"`. ## Attributes `name: str` The table's name in the plans that read it. `assumes: str | None` What the batches rest on, or None. ## Examples A source reading new rows from a list, by an increasing id: ``` python import pyarrow as pa import sverify as sv from sverify.io import Batch, Source rows = [{"id": 1, "amount": 30}, {"id": 2, "amount": 5}] class Orders(Source): def read(self, since): new = [r for r in rows if since is None or r["id"] > since] if not new: return None return Batch(pa.Table.from_pylist(new), cursor=new[-1]["id"]) @sv.transform() def big_orders(orders): return orders.filter(orders.amount > 10) big = big_orders(Orders("orders", assumes="ids only grow")) big.refresh() rows.append({"id": 3, "amount": 50}) print(big.refresh()) # big_orders: # orders@v1→v2: 1 insert(s), 0 delete(s); Δ trusted as rows appended # to orders v1→v2, assuming ids only grow # view big_orders: orders@v2 ``` ## Methods | Name | Description | |----|----| | [changes()](#changes) | The change from version `since` to the latest, as a refresh reads | | [latest()](#latest) | Read what is new, and return the latest version. | | [read()](#read) | What is new since the last read. | | [reread()](#reread) | The whole table as of `cursor`, read again from scratch, for | | [restore()](#restore) | The whole table as of `cursor`, a cursor that a sink recorded in | | [schema()](#schema) | The table's schema: `schema=` without reading, else the first | | [snapshot()](#snapshot) | The table as of a version: the latest read, or an earlier | ------------------------------------------------------------------------ #### changes() The change from version `since` to the latest, as a refresh reads Usage ``` python changes(since, until=None) ``` it: the batches read in between. ##### Parameters `since: int` The version the node's state was computed from. `until: int | None = None` The latest version, or None for it. ##### Returns `SourceBatch` The rows the batches added and removed, or, if they are no longer kept, the latest table. ##### Raises `ValueError` `until` is not the latest version. ------------------------------------------------------------------------ #### latest() Read what is new, and return the latest version. Usage ``` python latest() ``` ------------------------------------------------------------------------ #### read() What is new since the last read. Usage ``` python read(since) ``` ##### Parameters `since: Any` The cursor of the last batch read, or None on the first read (and after a batch whose cursor is None). ##### Returns `Batch | None` The new rows ([Batch](io.Batch.md#sverify.io.Batch)), or None if nothing is new. ------------------------------------------------------------------------ #### reread() The whole table as of `cursor`, read again from scratch, for Usage ``` python reread(cursor) ``` `refresh(audit=True)`. Override it to read only up to `cursor`, so rows that arrive meanwhile are not reported as a mismatch. ##### Parameters `cursor: Any` The cursor of the latest batch. ##### Returns `Any` The rows, in any form `Batch.table` takes. Default: the table of `read(None)`, or no rows. ------------------------------------------------------------------------ #### restore() The whole table as of `cursor`, a cursor that a sink recorded in Usage ``` python restore(cursor) ``` an earlier process, so a restarted job resumes from its sinks. Override it for a source whose cursors last: a log's sequence number, a high-water mark, the files read. Then `read(cursor)` must return what is new since, and the next refresh is incremental. A sink records the cursor if it has a JSON form (see `Batch.cursor`). The restored table is trusted to be the table the stored nodes were computed from; `refresh(audit=True)` reads the whole table again on later batches. ##### Parameters `cursor: Any` The cursor the sink recorded, as [read](odcs.read.md#sverify.odcs.read) returned it (datetimes, dates, decimals, bytes and tuples come back as they were). ##### Returns `Any` The rows, in any form `Batch.table` takes, or None if the cursor is no longer valid (the rows it points to are gone): then the job reads the whole table, `read(None)`, and recomputes. `Default: Any` None. ------------------------------------------------------------------------ #### schema() The table's schema: `schema=` without reading, else the first Usage ``` python schema() ``` batch's. ------------------------------------------------------------------------ #### snapshot() The table as of a version: the latest read, or an earlier Usage ``` python snapshot(version=None) ``` version whose batches the source still keeps. ##### Parameters `version: int | None = None` The version, or None for the latest. ##### Returns `Snapshot` The version and the rows, with no facts. ##### Raises `ValueError` The source no longer keeps `version`'s batches, or `version` is newer than the latest.