Airbyte is an open-source data integration engine that helps you consolidate your data in your data warehouses, lakes and databases.

Airbyte

Does it make sense to filter items in `read_records` if source API does not support query params to get incremental new or modified data and use like an incremental stream?
```    def read_records(...) -&gt; Iterable[Mapping[str, Any]]:
        filtered = []
        items = super().read_records(sync_mode=sync_mode, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state)
        for item in items:
            if self._field_to_datetime(item[self.cursor_field]) &gt; self._field_to_datetime(stream_state[self.cursor_field]):
                filtered.append(item)
        yield from filtered```

it does! that’s a good way of implementing this

although I would recommend doing it without accumulating objects in memory e.g:

```    def read_records(...) -&gt; Iterable[Mapping[str, Any]]:
        filtered = []
        items = super().read_records(sync_mode=sync_mode, cursor_field=cursor_field, stream_slice=stream_slice, stream_state=stream_state)
        for item in items:
            if self._field_to_datetime(item[self.cursor_field]) &gt; self._field_to_datetime(stream_state[self.cursor_field]):
                yield item```