AirbyteLib: Add Lazy Datasets and iterator syntax support for datasets, caches, and read results

[aaronsteers]

Given a source and cache declared something like this...

    source = ab.get_connector("source-test", config={"apiKey": "test"})
    result: ReadResult = source.read(ab.new_local_cache())

This adds support for all of these syntaxes:

    # Access by key:
    cached_dataset: CachedDataset = result[stream_name]
    cached_dataset: CachedDataset = result.streams[stream_name]
    cached_dataset: CachedDataset = result.cache[stream_name]
    cached_dataset: CachedDataset = result.cache.streams[stream_name]

    # Check if key exists:
    assert stream_name in cached_dataset
    assert stream_name in cached_dataset.streams
    assert stream_name in cached_dataset.cache
    assert stream_name in cached_dataset.cache.streams

Above, the .streams property exists as a more explicit reference than calling iterator or key operations directly against the object.

CachedDatasets and SQLDataset now also have a "with_filter()" method that creates a new SQLDataset by filtering down the original dataset. This is done lazily, so nothing is done on the connection until the dataset is invokes as an iterator or its data is attempted to be read.

These are all equivalent:

    filtered_dataset: SQLDataset = cached_dataset.with_filter("column2 == 1")
    filtered_dataset: SQLDataset = cached_dataset.with_filter(text("column2 == 1"))
    filtered_dataset: SQLDataset = cached_dataset.with_filter(column("column2") == 1)

Other points:

• SQLDataset objects (the base class) do not support to_sql_table() but CachedDataset objects do.
• Filter can also be chained together:
  • filtered_dataset: SQLDataset = cached_dataset.with_filter("column2 == 1").with_filter("column1 == 'value1'")
• Internally, processors (which includes caches and file writers) now internally track which streams they've seen in self._streams_with_data. This is a subset of streams that are declared in the catalog, and this is also where a cache might contain different members than a read result - even if the underlying data is the same.

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

1 Ignored Deployment

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	⬜️ Ignored (Inspect)	Visit Preview		Jan 23, 2024 7:26pm

[aaronsteers]

This is new. Basically, this tells us all the streams we've seen so far.

[aaronsteers]

Per Joe, does this work?:

for stream_name, dataset in result.streams.items():
    ...

Update:

@flash1293 - Confirmed this syntax works as expected, and I've added tests that confirm this.

In the process I noticed a look-alike Source.streams: str which I've refactored as Source._selected_stream_names to avoid conflating with the behavior that treats Cache.streams and ReadResult.streams as (basically) Mapping[str, CachedDataset].