✨feat(source-sftp-bulk): adjust file record message protocol (#57514)

Author: aldogonzalez8

airbyte-integrations/connectors/source-sftp-bulk/integration_tests/integration_test.py

@@ -14,6 +14,7 @@
 
 from airbyte_cdk import AirbyteTracedException, ConfiguredAirbyteCatalog
 from airbyte_cdk.models import (
+ AirbyteRecordMessageFileReference,
  FailureType,
  Status,
 )
@@ -101,27 +102,37 @@ def test_get_files_empty_files(configured_catalog: ConfiguredAirbyteCatalog, con
 
 
 @pytest.mark.slow
-@pytest.mark.limit_memory("10 MB")
+@pytest.mark.limit_memory("11 MB")
 def test_get_file_csv_file_transfer(configured_catalog: ConfiguredAirbyteCatalog, config_fixture_use_file_transfer: Mapping[str, Any]):
  source = SourceSFTPBulk(catalog=configured_catalog, config=config_fixture_use_file_transfer, state=None)
  output = read(source=source, config=config_fixture_use_file_transfer, catalog=configured_catalog)
- expected_file_data = {
+
+ file_folder = "files/file_transfer"
+ file_name = "file_transfer_1.csv"
+ source_file_relative_path = f"{file_folder}/{file_name}"
+ expected_file_data = AirbyteRecordMessageFileReference(
+ file_size_bytes=46_754_266,
+ source_file_relative_path=source_file_relative_path,
+ staging_file_url=f"/tmp/airbyte-file-transfer/{source_file_relative_path}",
+ )
+ expected_record_data = {
  "bytes": 46_754_266,
- "file_relative_path": "files/file_transfer/file_transfer_1.csv",
- "file_url": "/tmp/airbyte-file-transfer/files/file_transfer/file_transfer_1.csv",
- "modified": ANY,
- "source_file_url": "/files/file_transfer/file_transfer_1.csv",
+ "file_name": file_name,
+ "folder": f"/{file_folder}",
+ "source_uri": f"sftp://{config_fixture_use_file_transfer['username']}@{config_fixture_use_file_transfer['host']}:{config_fixture_use_file_transfer['port']}/{source_file_relative_path}",
+ "updated_at": ANY,
  }
  assert len(output.records) == 1
- assert list(map(lambda record: record.record.file, output.records)) == [expected_file_data]
+ assert list(map(lambda record: record.record.file_reference, output.records)) == [expected_file_data]
+ assert list(map(lambda record: record.record.data, output.records)) == [expected_record_data]
 
  # Additional assertion to check if the file exists at the file_url path
- file_path = expected_file_data["file_url"]
+ file_path = expected_file_data.staging_file_url
  assert os.path.exists(file_path), f"File not found at path: {file_path}"
 
 
 @pytest.mark.slow
-@pytest.mark.limit_memory("10 MB")
+@pytest.mark.limit_memory("11 MB")
 def test_get_all_file_csv_file_transfer(
  configured_catalog: ConfiguredAirbyteCatalog, config_fixture_use_all_files_transfer: Mapping[str, Any]
 ):
@@ -132,8 +143,8 @@ def test_get_all_file_csv_file_transfer(
  source = SourceSFTPBulk(catalog=configured_catalog, config=config_fixture_use_all_files_transfer, state=None)
  output = read(source=source, config=config_fixture_use_all_files_transfer, catalog=configured_catalog)
  assert len(output.records) == 5
- total_bytes = sum(list(map(lambda record: record.record.file["bytes"], output.records)))
- files_paths = list(map(lambda record: record.record.file["file_url"], output.records))
+ total_bytes = sum(list(map(lambda record: record.record.file_reference.file_size_bytes, output.records)))
+ files_paths = list(map(lambda record: record.record.file_reference.staging_file_url, output.records))
  for file_path in files_paths:
  assert os.path.exists(file_path), f"File not found at path: {file_path}"
  assert total_bytes == 233_771_330
@@ -150,8 +161,8 @@ def test_default_mirroring_paths_works_for_not_present_config_file_transfer(
  source = SourceSFTPBulk(catalog=configured_catalog, config=config_fixture_not_duplicates, state=None)
  output = read(source=source, config=config_fixture_not_duplicates, catalog=configured_catalog)
  assert len(output.records) == expected_uniqueness_count
- files_paths = set(map(lambda record: record.record.file["file_url"], output.records))
- files_relative_paths = set(map(lambda record: record.record.file["file_relative_path"], output.records))
+ files_paths = set(map(lambda record: record.record.file_reference.staging_file_url, output.records))
+ files_relative_paths = set(map(lambda record: record.record.file_reference.source_file_relative_path, output.records))
  assert len(files_relative_paths) == expected_uniqueness_count
  assert len(files_paths) == expected_uniqueness_count
  for file_path, files_relative_path in zip(files_paths, files_relative_paths):
@@ -170,8 +181,8 @@ def test_not_mirroring_paths_not_duplicates_file_transfer(
  source = SourceSFTPBulk(catalog=configured_catalog, config=config_fixture_not_mirroring_paths_not_duplicates, state=None)
  output = read(source=source, config=config_fixture_not_mirroring_paths_not_duplicates, catalog=configured_catalog)
  assert len(output.records) == expected_uniqueness_count
- files_paths = set(map(lambda record: record.record.file["file_url"], output.records))
- files_relative_paths = set(map(lambda record: record.record.file["file_relative_path"], output.records))
+ files_paths = set(map(lambda record: record.record.file_reference.staging_file_url, output.records))
+ files_relative_paths = set(map(lambda record: record.record.file_reference.source_file_relative_path, output.records))
  assert len(files_relative_paths) == expected_uniqueness_count
  assert len(files_paths) == expected_uniqueness_count
  for file_path, files_relative_path in zip(files_paths, files_relative_paths):

airbyte-integrations/connectors/source-sftp-bulk/metadata.yaml

@@ -7,7 +7,7 @@ data:
  connectorSubtype: file
  connectorType: source
  definitionId: 31e3242f-dee7-4cdc-a4b8-8e06c5458517
- dockerImageTag: 1.7.8
+ dockerImageTag: 1.8.0
  dockerRepository: airbyte/source-sftp-bulk
  documentationUrl: https://docs.airbyte.com/integrations/sources/sftp-bulk
  githubIssueLabel: source-sftp-bulk