✨Source Airtable: add table name to data records (airbytehq#31044)

airbyte-integrations/connectors/source-airtable/Dockerfile

@@ -34,5 +34,5 @@ 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=4.0.0
+LABEL io.airbyte.version=4.1.0
 LABEL io.airbyte.name=airbyte/source-airtable

airbyte-integrations/connectors/source-airtable/metadata.yaml

@@ -6,7 +6,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: 14c6e7ea-97ed-4f5e-a7b5-25e9a80b8212
-  dockerImageTag: 4.0.0
+  dockerImageTag: 4.1.0
   dockerRepository: airbyte/source-airtable
   githubIssueLabel: source-airtable
   icon: airtable.svg

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

@@ -63,7 +63,7 @@ class SchemaTypes:
     "singleSelect": SchemaTypes.string,
     "externalSyncSource": SchemaTypes.string,
     "url": SchemaTypes.string,
-    # referal default type
+    # referral default type
     "simpleText": SchemaTypes.string,
 }
 
@@ -89,6 +89,7 @@ def get_json_schema(table: Dict[str, Any]) -> Dict[str, str]:
         properties: Dict = {
             "_airtable_id": SchemaTypes.string,
             "_airtable_created_time": SchemaTypes.string,
+            "_airtable_table_name": SchemaTypes.string,
         }
 
         fields: Dict = table.get("fields", {})

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

@@ -86,6 +86,7 @@ def discover(self, logger: AirbyteLogger, config) -> AirbyteCatalog:
                             f"{base_name}/{SchemaHelpers.clean_name(table.get('name'))}/{table.get('id')}",
                             SchemaHelpers.get_json_schema(table),
                         ),
+                        "table_name": table.get("name"),
                     }
                 )
         return AirbyteCatalog(streams=[stream["stream"] for stream in self.streams_catalog])
@@ -101,5 +102,6 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
                 stream_path=stream["stream_path"],
                 stream_name=stream["stream"].name,
                 stream_schema=stream["stream"].json_schema,
+                table_name=stream["table_name"],
                 authenticator=self._auth,
             )

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

@@ -89,11 +89,12 @@ def path(self, **kwargs) -> str:
 
 
 class AirtableStream(HttpStream, ABC):
-    def __init__(self, stream_path: str, stream_name: str, stream_schema, **kwargs):
+    def __init__(self, stream_path: str, stream_name: str, stream_schema, table_name: str, **kwargs):
         super().__init__(**kwargs)
         self.stream_path = stream_path
         self.stream_name = stream_name
         self.stream_schema = stream_schema
+        self.table_name = table_name
 
     url_base = URL_BASE
     primary_key = "id"
@@ -146,6 +147,7 @@ def process_records(self, records) -> Iterable[Mapping[str, Any]]:
                 yield {
                     "_airtable_id": record.get("id"),
                     "_airtable_created_time": record.get("createdTime"),
+                    "_airtable_table_name": self.table_name,
                     **{SchemaHelpers.clean_name(k): v for k, v in data.items()},
                 }
 

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

@@ -89,11 +89,12 @@ def streams_json_response():
 
 
 @pytest.fixture
-def streams_processed_response():
+def streams_processed_response(table):
     return [
         {
             "_airtable_id": "some_id",
             "_airtable_created_time": "2022-12-02T19:50:00.000Z",
+            "_airtable_table_name": table,
             "field1": True,
             "field2": "test",
             "field3": 123,
@@ -109,14 +110,15 @@ def expected_json_schema():
         "properties": {
             "_airtable_created_time": {"type": ["null", "string"]},
             "_airtable_id": {"type": ["null", "string"]},
+            "_airtable_table_name": {"type": ["null", "string"]},
             "test": {"type": ["null", "string"]},
         },
         "type": "object",
     }
 
 
 @pytest.fixture(scope="function", autouse=True)
-def prepared_stream():
+def prepared_stream(table):
     return {
         "stream_path": "some_base_id/some_table_id",
         "stream": AirbyteStream(
@@ -128,12 +130,14 @@ def prepared_stream():
                 "properties": {
                     "_airtable_id": {"type": ["null", "string"]},
                     "_airtable_created_time": {"type": ["null", "string"]},
+                    "_airtable_table_name": {"type": ["null", "string"]},
                     "name": {"type": ["null", "string"]},
                 },
             },
             supported_sync_modes=[SyncMode.full_refresh],
             supported_destination_sync_modes=[DestinationSyncMode.overwrite, DestinationSyncMode.append_dedup],
         ),
+        "table_name": table,
     }
 
 
@@ -144,6 +148,7 @@ def make(name):
             stream_path=prepared_stream["stream_path"],
             stream_name=name,
             stream_schema=prepared_stream["stream"].json_schema,
+            table_name=prepared_stream["table_name"],
             authenticator=fake_auth,
         )
 

airbyte-integrations/connectors/source-airtable/unit_tests/expected_schema_for_sample_table.json

@@ -8,6 +8,9 @@
     "_airtable_id": {
       "type": ["null", "string"]
     },
+    "_airtable_table_name": {
+      "type": ["null", "string"]
+    },
     "assignee_(from_table_6)": {
       "items": {
         "type": ["null", "number"]

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

@@ -1,6 +1,7 @@
 #
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
+
 import json
 import os
 from typing import Any, Mapping

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

@@ -93,6 +93,7 @@ def stream_instance(self, prepared_stream):
             stream_path=prepared_stream["stream_path"],
             stream_name=prepared_stream["stream"].name,
             stream_schema=prepared_stream["stream"].json_schema,
+            table_name=prepared_stream["table_name"],
             authenticator=MagicMock(),
         )
 

docs/integrations/sources/airtable.md

@@ -111,6 +111,7 @@ See information about rate limits [here](https://airtable.com/developers/web/api
 
 | Version | Date       | Pull Request                                             | Subject                                                                                |
 |:--------|:-----------|:---------------------------------------------------------|:---------------------------------------------------------------------------------------|
+| 4.1.0   | 2023-10-10 | [31044](https://github.com/airbytehq/airbyte/pull/31044) | Add source table name to output records    |
 | 4.0.0   | 2023-10-09 | [31181](https://github.com/airbytehq/airbyte/pull/31181) | Additional schema processing for the FORMULA schema type: Convert to simple data types |
 | 3.0.1   | 2023-05-10 | [25946](https://github.com/airbytehq/airbyte/pull/25946) | Skip stream if it does not appear in catalog                                           |
 | 3.0.0   | 2023-03-20 | [22704](https://github.com/airbytehq/airbyte/pull/22704) | Fix for stream name uniqueness                                                         |