✨Source Airtable: add table name to data records

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,
 }
 
@@ -87,6 +87,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/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(),
         )