🎉 New Source: Airtable (#7639)

airbyte-integrations/connectors/source-airtable/Dockerfile

@@ -0,0 +1,38 @@
+FROM python:3.7.11-alpine3.14 as base
+
+# build and load all requirements
+FROM base as builder
+WORKDIR /airbyte/integration_code
+
+# upgrade pip to the latest version
+RUN apk --no-cache upgrade \
+    && pip install --upgrade pip \
+    && apk --no-cache add tzdata build-base
+
+
+COPY setup.py ./
+# install necessary packages to a temporary folder
+RUN pip install --prefix=/install .
+
+# build a clean environment
+FROM base
+WORKDIR /airbyte/integration_code
+
+# copy all loaded and built libraries to a pure basic image
+COPY --from=builder /install /usr/local
+# add default timezone settings
+COPY --from=builder /usr/share/zoneinfo/Etc/UTC /etc/localtime
+RUN echo "Etc/UTC" > /etc/timezone
+
+# bash is installed for more convenient debugging.
+RUN apk --no-cache add bash
+
+# copy payload code only
+COPY main.py ./
+COPY source_airtable ./source_airtable
+
+ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
+ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
+
+LABEL io.airbyte.version=0.1.0
+LABEL io.airbyte.name=airbyte/source-airtable

airbyte-integrations/connectors/source-airtable/README.md

@@ -0,0 +1,133 @@
+# Airtable Source
+
+This is the repository for the Airtable source connector, written in Python.
+For information about how to use this connector within Airbyte, see [the documentation](https://docs.airbyte.io/integrations/sources/airtable).
+
+## Local development
+
+### Prerequisites
+**To iterate on this connector, make sure to complete this prerequisites section.**
+- Create a base named `users` in your AirTable account.
+- Create two tables named `Table 1` and `Table 2` in the `users` base.
+#### Minimum Python version required `= 3.7.0`
+
+#### Build & Activate Virtual Environment and install dependencies
+From this connector directory, create a virtual environment:
+```
+python -m venv .venv
+```
+
+This will generate a virtualenv for this module in `.venv/`. Make sure this venv is active in your
+development environment of choice. To activate it from the terminal, run:
+```
+source .venv/bin/activate
+pip install -r requirements.txt
+pip install '.[tests]'
+```
+If you are in an IDE, follow your IDE's instructions to activate the virtualenv.
+
+Note that while we are installing dependencies from `requirements.txt`, you should only edit `setup.py` for your dependencies. `requirements.txt` is
+used for editable installs (`pip install -e`) to pull in Python dependencies from the monorepo and will call `setup.py`.
+If this is mumbo jumbo to you, don't worry about it, just put your deps in `setup.py` but install using `pip install -r requirements.txt` and everything
+should work as you expect.
+
+#### Building via Gradle
+You can also build the connector in Gradle. This is typically used in CI and not needed for your development workflow.
+
+To build using Gradle, from the Airbyte repository root, run:
+```
+./gradlew :airbyte-integrations:connectors:source-airtable:build
+```
+
+#### Create credentials
+**If you are a community contributor**, follow the instructions in the [documentation](https://docs.airbyte.io/integrations/sources/airtable)
+to generate the necessary credentials. Then create a file `secrets/config.json` conforming to the `source_airtable/spec.json` file.
+Note that any directory named `secrets` is gitignored across the entire Airbyte repo, so there is no danger of accidentally checking in sensitive information.
+See `integration_tests/sample_config.json` for a sample config file.
+
+**If you are an Airbyte core member**, copy the credentials in Lastpass under the secret name `source airtable test creds`
+and place them into `secrets/config.json`.
+
+### Locally running the connector
+```
+python main.py spec
+python main.py check --config secrets/config.json
+python main.py discover --config secrets/config.json
+python main.py read --config secrets/config.json --catalog integration_tests/configured_catalog.json
+```
+
+### Locally running the connector docker image
+
+#### Build
+First, make sure you build the latest Docker image:
+```
+docker build . -t airbyte/source-airtable:dev
+```
+
+You can also build the connector image via Gradle:
+```
+./gradlew :airbyte-integrations:connectors:source-airtable:airbyteDocker
+```
+When building via Gradle, the docker image name and tag, respectively, are the values of the `io.airbyte.name` and `io.airbyte.version` `LABEL`s in
+the Dockerfile.
+
+#### Run
+Then run any of the connector commands as follows:
+```
+docker run --rm airbyte/source-airtable:dev spec
+docker run --rm -v $(pwd)/secrets:/secrets airbyte/source-airtable:dev check --config /secrets/config.json
+docker run --rm -v $(pwd)/secrets:/secrets airbyte/source-airtable:dev discover --config /secrets/config.json
+docker run --rm -v $(pwd)/secrets:/secrets -v $(pwd)/integration_tests:/integration_tests airbyte/source-airtable:dev read --config /secrets/config.json --catalog /integration_tests/configured_catalog.json
+```
+## Testing
+Make sure to familiarize yourself with [pytest test discovery](https://docs.pytest.org/en/latest/goodpractices.html#test-discovery) to know how your test files and methods should be named.
+First install test dependencies into your virtual environment:
+```
+pip install .[tests]
+```
+### Unit Tests
+To run unit tests locally, from the connector directory run:
+```
+python -m pytest unit_tests
+```
+
+### Integration Tests
+There are two types of integration tests: Acceptance Tests (Airbyte's test suite for all source connectors) and custom integration tests (which are specific to this connector).
+#### Custom Integration tests
+Place custom tests inside `integration_tests/` folder, then, from the connector root, run
+```
+python -m pytest integration_tests
+```
+#### Acceptance Tests
+Customize `acceptance-test-config.yml` file to configure tests. See [Source Acceptance Tests](https://docs.airbyte.io/connector-development/testing-connectors/source-acceptance-tests-reference) for more information.
+If your connector requires to create or destroy resources for use during acceptance tests create fixtures for it and place them inside integration_tests/acceptance.py.
+To run your integration tests with acceptance tests, from the connector root, run
+```
+python -m pytest integration_tests -p integration_tests.acceptance
+```
+To run your integration tests with docker
+
+### Using gradle to run tests
+All commands should be run from airbyte project root.
+To run unit tests:
+```
+./gradlew :airbyte-integrations:connectors:source-airtable:unitTest
+```
+To run acceptance and custom integration tests:
+```
+./gradlew :airbyte-integrations:connectors:source-airtable:integrationTest
+```
+
+## Dependency Management
+All of your dependencies should go in `setup.py`, NOT `requirements.txt`. The requirements file is only used to connect internal Airbyte dependencies in the monorepo for local development.
+We split dependencies between two groups, dependencies that are:
+* required for your connector to work need to go to `MAIN_REQUIREMENTS` list.
+* required for the testing need to go to `TEST_REQUIREMENTS` list
+
+### Publishing a new version of the connector
+You've checked out the repo, implemented a million dollar feature, and you're ready to share your changes with the world. Now what?
+1. Make sure your changes are passing unit and integration tests.
+1. Bump the connector version in `Dockerfile` -- just increment the value of the `LABEL io.airbyte.version` appropriately (we use [SemVer](https://semver.org/)).
+1. Create a Pull Request.
+1. Pat yourself on the back for being an awesome contributor.
+1. Someone from Airbyte will take a look at your PR and iterate with you to merge it into master.

airbyte-integrations/connectors/source-airtable/acceptance-test-config.yml

@@ -0,0 +1,20 @@
+# See [Source Acceptance Tests](https://docs.airbyte.io/connector-development/testing-connectors/source-acceptance-tests-reference)
+# for more information about how to configure these tests
+connector_image: airbyte/source-airtable:dev
+tests:
+  spec:
+    - spec_path: "source_airtable/spec.json"
+  connection:
+    - config_path: "secrets/config.json"
+      status: "succeed"
+    - config_path: "integration_tests/invalid_config.json"
+      status: "failed"
+  discovery:
+    - config_path: "secrets/config.json"
+  basic_read:
+    - config_path: "secrets/config.json"
+      configured_catalog_path: "integration_tests/configured_catalog.json"
+      empty_streams: []
+  full_refresh:
+    - config_path: "secrets/config.json"
+      configured_catalog_path: "integration_tests/configured_catalog.json"

airbyte-integrations/connectors/source-airtable/acceptance-test-docker.sh

@@ -0,0 +1,16 @@
+#!/usr/bin/env sh
+
+# Build latest connector image
+docker build . -t $(cat acceptance-test-config.yml | grep "connector_image" | head -n 1 | cut -d: -f2)
+
+# Pull latest acctest image
+docker pull airbyte/source-acceptance-test:latest
+
+# Run
+docker run --rm -it \
+    -v /var/run/docker.sock:/var/run/docker.sock \
+    -v /tmp:/tmp \
+    -v $(pwd):/test_input \
+    airbyte/source-acceptance-test \
+    --acceptance-test-config /test_input
+

airbyte-integrations/connectors/source-airtable/build.gradle

@@ -0,0 +1,14 @@
+plugins {
+    id 'airbyte-python'
+    id 'airbyte-docker'
+    id 'airbyte-source-acceptance-test'
+}
+
+airbytePython {
+    moduleDirectory 'source_airtable'
+}
+
+dependencies {
+    implementation files(project(':airbyte-integrations:bases:source-acceptance-test').airbyteDocker.outputs)
+    implementation files(project(':airbyte-integrations:bases:base-python').airbyteDocker.outputs)
+}

airbyte-integrations/connectors/source-airtable/integration_tests/__init__.py

@@ -0,0 +1,3 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#

airbyte-integrations/connectors/source-airtable/integration_tests/acceptance.py

@@ -0,0 +1,14 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+
+import pytest
+
+pytest_plugins = ("source_acceptance_test.plugin",)
+
+
+@pytest.fixture(scope="session", autouse=True)
+def connector_setup():
+    """This fixture is a placeholder for external resources that acceptance test might require."""
+    yield

airbyte-integrations/connectors/source-airtable/integration_tests/configured_catalog.json

@@ -0,0 +1,32 @@
+{
+  "streams": [
+    {
+      "stream": {
+        "name": "Table 1",
+        "json_schema": {
+          "$schema": "http://json-schema.org/draft-07/schema#",
+          "type": "object",
+          "properties": {}
+        },
+        "supported_sync_modes": ["full_refresh"],
+        "supported_destination_sync_modes": ["overwrite", "append_dedup"]
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "overwrite"
+    },
+    {
+      "stream": {
+        "name": "Table 2",
+        "json_schema": {
+          "$schema": "http://json-schema.org/draft-07/schema#",
+          "type": "object",
+          "properties": {}
+        },
+        "supported_sync_modes": ["full_refresh"],
+        "supported_destination_sync_modes": ["overwrite", "append_dedup"]
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "overwrite"
+    }
+  ]
+}

airbyte-integrations/connectors/source-airtable/integration_tests/invalid_config.json

@@ -0,0 +1,5 @@
+{
+  "api_key": "key####################",
+  "base_id": "app####################",
+  "tables": ["Table 1", "Table 2"]
+}

airbyte-integrations/connectors/source-airtable/integration_tests/sample_config.json

@@ -0,0 +1,5 @@
+{
+  "api_key": "key1234567890",
+  "base_id": "app1234567890",
+  "tables": ["Table 1", "Table 2"]
+}

airbyte-integrations/connectors/source-airtable/main.py

@@ -0,0 +1,13 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+
+import sys
+
+from airbyte_cdk.entrypoint import launch
+from source_airtable import SourceAirtable
+
+if __name__ == "__main__":
+    source = SourceAirtable()
+    launch(source, sys.argv[1:])

airbyte-integrations/connectors/source-airtable/requirements.txt

@@ -0,0 +1,2 @@
+-e ../../bases/source-acceptance-test
+-e .

airbyte-integrations/connectors/source-airtable/setup.py

@@ -0,0 +1,29 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+
+from setuptools import find_packages, setup
+
+MAIN_REQUIREMENTS = [
+    "airbyte-cdk",
+]
+
+TEST_REQUIREMENTS = [
+    "pytest~=6.1",
+    "pytest-mock~=3.6.1",
+    "source-acceptance-test",
+]
+
+setup(
+    name="source_airtable",
+    description="Source implementation for Airtable.",
+    author="Airbyte",
+    author_email="anhtuan.nguyen@me.com",
+    packages=find_packages(),
+    install_requires=MAIN_REQUIREMENTS,
+    package_data={"": ["*.json", "schemas/*.json", "schemas/shared/*.json"]},
+    extras_require={
+        "tests": TEST_REQUIREMENTS,
+    },
+)

airbyte-integrations/connectors/source-airtable/source_airtable/__init__.py

@@ -0,0 +1,8 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+
+from .source import SourceAirtable
+
+__all__ = ["SourceAirtable"]

airbyte-integrations/connectors/source-airtable/source_airtable/helpers.py

@@ -0,0 +1,57 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+
+from typing import Any, Dict
+
+import requests
+from airbyte_cdk.models import AirbyteStream
+from airbyte_cdk.models.airbyte_protocol import DestinationSyncMode, SyncMode
+from airbyte_cdk.sources.streams.http.requests_native_auth import TokenAuthenticator
+
+
+class Helpers(object):
+    @staticmethod
+    def get_first_row(auth: TokenAuthenticator, base_id: str, table: str) -> Dict[str, Any]:
+        url = f"https://api.airtable.com/v0/{base_id}/{table}?pageSize=1"
+        try:
+            response = requests.get(url, headers=auth.get_auth_header())
+            response.raise_for_status()
+        except requests.exceptions.HTTPError as e:
+            if e.response.status_code == 401:
+                raise Exception("Invalid API key")
+            elif e.response.status_code == 404:
+                raise Exception(f"Table '{table}' not found")
+            else:
+                raise Exception(f"Error getting first row from table {table}: {e}")
+        json_response = response.json()
+        record = json_response.get("records", [])[0]
+        return record
+
+    @staticmethod
+    def get_json_schema(record: Dict[str, Any]) -> Dict[str, str]:
+        fields = record.get("fields", {})
+        properties = {
+            "_airtable_id": {"type": ["null", "string"]},
+            "_airtable_created_time": {"type": ["null", "string"]},
+        }
+
+        for field in fields:
+            properties[field] = {"type": ["null", "string"]}
+
+        json_schema = {
+            "$schema": "http://json-schema.org/draft-07/schema#",
+            "type": "object",
+            "properties": properties,
+        }
+        return json_schema
+
+    @staticmethod
+    def get_airbyte_stream(table: str, json_schema: Dict[str, Any]) -> AirbyteStream:
+        return AirbyteStream(
+            name=table,
+            json_schema=json_schema,
+            supported_sync_modes=[SyncMode.full_refresh],
+            supported_destination_sync_modes=[DestinationSyncMode.overwrite, DestinationSyncMode.append_dedup],
+        )

airbyte-integrations/connectors/source-airtable/source_airtable/source.py

@@ -0,0 +1,101 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+
+from abc import ABC
+from typing import Any, Iterable, List, Mapping, MutableMapping, Optional, Tuple
+
+import requests
+from airbyte_cdk.logger import AirbyteLogger
+from airbyte_cdk.models import AirbyteCatalog
+from airbyte_cdk.sources import AbstractSource
+from airbyte_cdk.sources.streams import Stream
+from airbyte_cdk.sources.streams.http import HttpStream
+from airbyte_cdk.sources.streams.http.requests_native_auth import TokenAuthenticator
+from airbyte_cdk.sources.utils.transform import TransformConfig, TypeTransformer
+
+from .helpers import Helpers
+
+
+# Basic full refresh stream
+class AirtableStream(HttpStream, ABC):
+    url_base = "https://api.airtable.com/v0/"
+    primary_key = "id"
+    transformer: TypeTransformer = TypeTransformer(TransformConfig.DefaultSchemaNormalization)
+
+    def __init__(self, base_id: str, table_name: str, schema, **kwargs):
+        super().__init__(**kwargs)
+        self.base_id = base_id
+        self.table_name = table_name
+        self.schema = schema
+
+    @property
+    def name(self):
+        return self.table_name
+
+    def get_json_schema(self) -> Mapping[str, Any]:
+        return self.schema
+
+    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
+        json_response = response.json()
+        offset = json_response.get("offset", None)
+        if offset:
+            return {"offset": offset}
+        return None
+
+    def request_params(
+        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
+    ) -> MutableMapping[str, Any]:
+        if next_page_token:
+            return next_page_token
+        return {}
+
+    def process_records(self, records):
+        for record in records:
+            data = record.get("fields", {})
+            processed_record = {"_airtable_id": record.get("id"), "_airtable_created_time": record.get("createdTime"), **data}
+            yield processed_record
+
+    def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapping]:
+        json_response = response.json()
+        records = json_response.get("records", [])
+        records = self.process_records(records)
+        yield from records
+
+    def path(
+        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
+    ) -> str:
+        return f"{self.base_id}/{self.table_name}"
+
+
+# Source
+class SourceAirtable(AbstractSource):
+    def check_connection(self, logger, config) -> Tuple[bool, any]:
+        auth = TokenAuthenticator(token=config["api_key"])
+        for table in config["tables"]:
+            try:
+                Helpers.get_first_row(auth, config["base_id"], table)
+            except Exception as e:
+                return False, str(e)
+        return True, None
+
+    def discover(self, logger: AirbyteLogger, config) -> AirbyteCatalog:
+        streams = []
+        auth = TokenAuthenticator(token=config["api_key"])
+        for table in config["tables"]:
+            record = Helpers.get_first_row(auth, config["base_id"], table)
+            json_schema = Helpers.get_json_schema(record)
+            airbyte_stream = Helpers.get_airbyte_stream(table, json_schema)
+            streams.append(airbyte_stream)
+        return AirbyteCatalog(streams=streams)
+
+    def streams(self, config: Mapping[str, Any]) -> List[Stream]:
+        auth = TokenAuthenticator(token=config["api_key"])
+        streams = []
+        for table in config["tables"]:
+            record = Helpers.get_first_row(auth, config["base_id"], table)
+            json_schema = Helpers.get_json_schema(record)
+            stream = AirtableStream(base_id=config["base_id"], table_name=table, authenticator=auth, schema=json_schema)
+            streams.append(stream)
+        return streams

airbyte-integrations/connectors/source-airtable/source_airtable/spec.json

@@ -0,0 +1,28 @@
+{
+  "documentationUrl": "https://docs.airbyte.io/integrations/sources/airtable",
+  "connectionSpecification": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "title": "Airtable Source Spec",
+    "type": "object",
+    "required": ["api_key", "base_id", "tables"],
+    "additionalProperties": false,
+    "properties": {
+      "api_key": {
+        "type": "string",
+        "description": "The API key for the Airtable account. https://support.airtable.com/hc/en-us/articles/219046777-How-do-I-get-my-API-key-",
+        "airbyte_secret": true,
+        "examples": ["key1234567890"]
+      },
+      "base_id": {
+        "type": "string",
+        "description": "The base ID to integrate data from",
+        "examples": ["app1234567890"]
+      },
+      "tables": {
+        "type": "array",
+        "description": "The list of tables to integrate",
+        "examples": ["table 1", "table 2"]
+      }
+    }
+  }
+}

airbyte-integrations/connectors/source-airtable/unit_tests/conftest.py

@@ -0,0 +1,10 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+import pytest
+
+
+@pytest.fixture
+def config():
+    return {"api_key": "key1234567890", "base_id": "app1234567890", "tables": ["Table 1", "Table 2"]}

airbyte-integrations/connectors/source-airtable/unit_tests/test_helpers.py

@@ -0,0 +1,80 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+from http import HTTPStatus
+from unittest.mock import MagicMock, patch
+
+import pytest
+from airbyte_cdk.sources.streams.http.requests_native_auth import TokenAuthenticator
+from source_airtable.helpers import Helpers
+
+
+@pytest.fixture
+def base_id():
+    return "app1234567890"
+
+
+@pytest.fixture
+def api_key():
+    return "key1234567890"
+
+
+@pytest.fixture
+def table():
+    return "Table 1"
+
+
+@pytest.fixture
+def auth():
+    return MagicMock()
+
+
+@pytest.fixture
+def json_response():
+    return {"records": [{"id": "abc", "fields": {"name": "test"}}]}
+
+
+@pytest.fixture
+def expected_json_schema():
+    return {
+        "$schema": "http://json-schema.org/draft-07/schema#",
+        "properties": {
+            "_airtable_created_time": {"type": ["null", "string"]},
+            "_airtable_id": {"type": ["null", "string"]},
+            "name": {"type": ["null", "string"]},
+        },
+        "type": "object",
+    }
+
+
+def test_get_first_row(auth, base_id, table, json_response):
+    with patch("requests.get") as mock_get:
+        mock_get.return_value.status_code = HTTPStatus.OK
+        mock_get.return_value.json.return_value = json_response
+        assert Helpers.get_first_row(auth, base_id, table) == {"id": "abc", "fields": {"name": "test"}}
+
+
+def test_get_first_row_invalid_api_key(base_id, table):
+    with pytest.raises(Exception):
+        auth = TokenAuthenticator("invalid_api_key")
+        Helpers.get_first_row(auth, base_id, table)
+
+
+def test_get_first_row_table_not_found(auth, base_id, table):
+    with patch("requests.exceptions.HTTPError") as mock_get:
+        mock_get.return_value.status_code = HTTPStatus.NOT_FOUND
+        with pytest.raises(Exception):
+            Helpers.get_first_row(auth, base_id, table)
+
+
+def test_get_json_schema(json_response, expected_json_schema):
+    json_schema = Helpers.get_json_schema(json_response["records"][0])
+    assert json_schema == expected_json_schema
+
+
+def test_get_airbyte_stream(table, expected_json_schema):
+    stream = Helpers.get_airbyte_stream(table, expected_json_schema)
+    assert stream
+    assert stream.name == table
+    assert stream.json_schema == expected_json_schema

airbyte-integrations/connectors/source-airtable/unit_tests/test_source.py

@@ -0,0 +1,40 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+from unittest.mock import MagicMock, patch
+
+from airbyte_cdk.models import AirbyteCatalog, ConnectorSpecification
+from source_airtable.helpers import Helpers
+from source_airtable.source import SourceAirtable
+
+
+def test_spec(config):
+    source = SourceAirtable()
+    logger_mock = MagicMock()
+    spec = source.spec(logger_mock)
+    assert isinstance(spec, ConnectorSpecification)
+
+
+def test_discover(config, mocker):
+    source = SourceAirtable()
+    logger_mock, Helpers.get_first_row = MagicMock(), MagicMock()
+    airbyte_catalog = source.discover(logger_mock, config)
+    assert [stream.name for stream in airbyte_catalog.streams] == config["tables"]
+    assert isinstance(airbyte_catalog, AirbyteCatalog)
+    assert Helpers.get_first_row.call_count == 2
+
+
+@patch("requests.get")
+def test_check_connection(config):
+    source = SourceAirtable()
+    logger_mock = MagicMock()
+    assert source.check_connection(logger_mock, config) == (True, None)
+
+
+def test_streams(config):
+    source = SourceAirtable()
+    Helpers.get_first_row = MagicMock()
+    streams = source.streams(config)
+    assert len(streams) == 2
+    assert [stream.name for stream in streams] == config["tables"]