feat(source-sftp-bulk): not mirroring paths toggle for file transfer

airbyte-integrations/connectors/source-sftp-bulk/integration_tests/configs/config_not_duplicates.json

@@ -0,0 +1,59 @@
+{
+ "delivery_method": { "delivery_type": "use_file_transfer" },
+ "host": "localhost",
+ "port": 2222,
+ "username": "foo",
+ "credentials": {
+ "auth_type": "password",
+ "password": "pass"
+ },
+ "file_type": "json",
+ "start_date": "2021-01-01T00:00:00.000000Z",
+ "folder_path": "/files",
+ "streams": [
+ {
+ "name": "test_stream",
+ "file_type": "csv",
+ "globs": [
+ "**/monthly-kickoff-202401.mpeg",
+ "**/monthly-kickoff-202402.mpeg",
+ "**/monthly-kickoff-202403.mpeg"
+ ],
+ "legacy_prefix": "",
+ "validation_policy": "Emit Record",
+ "format": {
+ "filetype": "csv",
+ "delimiter": ",",
+ "quote_char": "\"",
+ "double_quote": true,
+ "null_values": [
+ "",
+ "#N/A",
+ "#N/A N/A",
+ "#NA",
+ "-1.#IND",
+ "-1.#QNAN",
+ "-NaN",
+ "-nan",
+ "1.#IND",
+ "1.#QNAN",
+ "N/A",
+ "NA",
+ "NULL",
+ "NaN",
+ "n/a",
+ "nan",
+ "null"
+ ],
+ "true_values": ["1", "True", "TRUE", "true"],
+ "false_values": ["0", "False", "FALSE", "false"],
+ "inference_type": "Primitive Types Only",
+ "strings_can_be_null": false,
+ "encoding": "utf8",
+ "header_definition": {
+ "header_definition_type": "From CSV"
+ }
+ }
+ }
+ ]
+}

airbyte-integrations/connectors/source-sftp-bulk/integration_tests/configs/config_not_preserve_subdirectories_not_duplicates.json

@@ -0,0 +1,62 @@
+{
+ "delivery_method": {
+ "delivery_type": "use_file_transfer",
+ "preserve_directory_structure": false
+ },
+ "host": "localhost",
+ "port": 2222,
+ "username": "foo",
+ "credentials": {
+ "auth_type": "password",
+ "password": "pass"
+ },
+ "file_type": "json",
+ "start_date": "2021-01-01T00:00:00.000000Z",
+ "folder_path": "/files",
+ "streams": [
+ {
+ "name": "test_stream",
+ "file_type": "csv",
+ "globs": [
+ "**/monthly-kickoff-202401.mpeg",
+ "**/monthly-kickoff-202402.mpeg",
+ "**/monthly-kickoff-202403.mpeg"
+ ],
+ "legacy_prefix": "",
+ "validation_policy": "Emit Record",
+ "format": {
+ "filetype": "csv",
+ "delimiter": ",",
+ "quote_char": "\"",
+ "double_quote": true,
+ "null_values": [
+ "",
+ "#N/A",
+ "#N/A N/A",
+ "#NA",
+ "-1.#IND",
+ "-1.#QNAN",
+ "-NaN",
+ "-nan",
+ "1.#IND",
+ "1.#QNAN",
+ "N/A",
+ "NA",
+ "NULL",
+ "NaN",
+ "n/a",
+ "nan",
+ "null"
+ ],
+ "true_values": ["1", "True", "TRUE", "true"],
+ "false_values": ["0", "False", "FALSE", "false"],
+ "inference_type": "Primitive Types Only",
+ "strings_can_be_null": false,
+ "encoding": "utf8",
+ "header_definition": {
+ "header_definition_type": "From CSV"
+ }
+ }
+ }
+ ]
+}

airbyte-integrations/connectors/source-sftp-bulk/integration_tests/configs/config_not_preserve_subdirectories_with_duplicates.json

@@ -0,0 +1,58 @@
+{
+ "delivery_method": {
+ "delivery_type": "use_file_transfer",
+ "preserve_directory_structure": false
+ },
+ "host": "localhost",
+ "port": 2222,
+ "username": "foo",
+ "credentials": {
+ "auth_type": "password",
+ "password": "pass"
+ },
+ "file_type": "json",
+ "start_date": "2021-01-01T00:00:00.000000Z",
+ "folder_path": "/files",
+ "streams": [
+ {
+ "name": "test_stream",
+ "file_type": "csv",
+ "globs": ["**/monthly-kickoff.mpeg"],
+ "legacy_prefix": "",
+ "validation_policy": "Emit Record",
+ "format": {
+ "filetype": "csv",
+ "delimiter": ",",
+ "quote_char": "\"",
+ "double_quote": true,
+ "null_values": [
+ "",
+ "#N/A",
+ "#N/A N/A",
+ "#NA",
+ "-1.#IND",
+ "-1.#QNAN",
+ "-NaN",
+ "-nan",
+ "1.#IND",
+ "1.#QNAN",
+ "N/A",
+ "NA",
+ "NULL",
+ "NaN",
+ "n/a",
+ "nan",
+ "null"
+ ],
+ "true_values": ["1", "True", "TRUE", "true"],
+ "false_values": ["0", "False", "FALSE", "false"],
+ "inference_type": "Primitive Types Only",
+ "strings_can_be_null": false,
+ "encoding": "utf8",
+ "header_definition": {
+ "header_definition_type": "From CSV"
+ }
+ }
+ }
+ ]
+}

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

@@ -124,6 +124,27 @@ def config_fixture_use_all_files_transfer(docker_client) -> Mapping[str, Any]:
  yield config
 
 
+@pytest.fixture(name="config_fixture_not_duplicates", scope="session")
+def config_fixture_not_duplicates(docker_client) -> Mapping[str, Any]:
+ config = load_config("config_not_duplicates.json")
+ config["host"] = get_docker_ip()
+ yield config
+
+
+@pytest.fixture(name="config_fixture_not_mirroring_paths_not_duplicates", scope="session")
+def config_fixture_not_mirroring_paths_not_duplicates(docker_client) -> Mapping[str, Any]:
+ config = load_config("config_not_preserve_subdirectories_not_duplicates.json")
+ config["host"] = get_docker_ip()
+ yield config
+
+
+@pytest.fixture(name="config_fixture_not_mirroring_paths_with_duplicates", scope="session")
+def config_fixture_not_mirroring_paths_with_duplicates(docker_client) -> Mapping[str, Any]:
+ config = load_config("config_not_preserve_subdirectories_with_duplicates.json")
+ config["host"] = get_docker_ip()
+ yield config
+
+
 @pytest.fixture(name="config_private_key", scope="session")
 def config_fixture_private_key(docker_client) -> Mapping[str, Any]:
  config = load_config("config_private_key.json") | {

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

@@ -12,7 +12,11 @@
 import pytest
 from source_sftp_bulk import SourceSFTPBulk
 
-from airbyte_cdk import AirbyteTracedException, ConfiguredAirbyteCatalog, Status
+from airbyte_cdk import AirbyteTracedException, ConfiguredAirbyteCatalog
+from airbyte_cdk.models import (
+ FailureType,
+ Status,
+)
 from airbyte_cdk.sources.declarative.models import FailureType
 from airbyte_cdk.test.entrypoint_wrapper import read
 
@@ -133,3 +137,62 @@ def test_get_all_file_csv_file_transfer(
  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
+
+
+def test_default_mirroring_paths_works_for_not_present_config_file_transfer(
+ configured_catalog: ConfiguredAirbyteCatalog, config_fixture_not_duplicates: Mapping[str, Any]
+):
+ """
+ If delivery_options is not provided in the config we fall back preserve directories (mirroring paths).
+ """
+ expected_directory_path = "files/not_duplicates/data/"
+ expected_uniqueness_count = 3
+ 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))
+ 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):
+ assert expected_directory_path in file_path, f"File not found at path: {file_path}"
+ assert expected_directory_path in files_relative_path, f"File not found at path: {files_relative_path}"
+
+
+def test_not_mirroring_paths_not_duplicates_file_transfer(
+ configured_catalog: ConfiguredAirbyteCatalog, config_fixture_not_mirroring_paths_not_duplicates: Mapping[str, Any]
+):
+ """
+ Delivery options is present and preserve_directory_structure is False so we should not preserve directories (mirroring paths).
+ """
+ source_directory_path = "files/not_duplicates/data/"
+ expected_uniqueness_count = 3
+ 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))
+ 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):
+ assert source_directory_path not in file_path, f"Source path found but mirroring is off: {file_path}"
+ assert source_directory_path not in files_relative_path, f"Source path found but mirroring is off: {files_relative_path}"
+
+
+def test_not_mirroring_paths_with_duplicates_file_transfer_fails_sync(
+ configured_catalog: ConfiguredAirbyteCatalog, config_fixture_not_mirroring_paths_with_duplicates: Mapping[str, Any]
+):
+ """
+ Delivery options is present and preserve_directory_structure is False so we should not preserve directories (mirroring paths),
+ but, there are duplicates so the sync fails.
+ """
+
+ source = SourceSFTPBulk(catalog=configured_catalog, config=config_fixture_not_mirroring_paths_with_duplicates, state=None)
+ output = read(source=source, config=config_fixture_not_mirroring_paths_with_duplicates, catalog=configured_catalog)
+
+ # assert error_message in output.errors[-1].trace.error.message
+ assert "3 duplicates found for file name monthly-kickoff.mpeg" in output.errors[-1].trace.error.message
+ assert "/files/duplicates/data/feb/monthly-kickoff.mpeg" in output.errors[-1].trace.error.message
+ assert "/files/duplicates/data/jan/monthly-kickoff.mpeg" in output.errors[-1].trace.error.message
+ assert "/files/duplicates/data/mar/monthly-kickoff.mpeg" in output.errors[-1].trace.error.message
+ assert not output.records

airbyte-integrations/connectors/source-sftp-bulk/integration_tests/spec.json

@@ -463,6 +463,12 @@
  "const": "use_file_transfer",
  "enum": ["use_file_transfer"],
  "type": "string"
+ },
+ "preserve_directory_structure": {
+ "default": true,
+ "description": "If enabled, sends subdirectory folder structure along with source file names to the destination. Otherwise, files will be synced by their names only. This option is ignored when file-based replication is not enabled.",
+ "title": "Preserve Sub-Directories in File Paths",
+ "type": "boolean"
  }
  },
  "description": "Copy raw files without parsing their contents. Bits are copied into the destination exactly as they appeared in the source. Recommended for use with unstructured text data, non-text and compressed files.",

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.6.0
+ dockerImageTag: 1.7.0-rc.1
  dockerRepository: airbyte/source-sftp-bulk
  documentationUrl: https://docs.airbyte.com/integrations/sources/sftp-bulk
  githubIssueLabel: source-sftp-bulk
@@ -25,6 +25,8 @@ data:
  enabled: true
  releaseStage: alpha
  releases:
+ rolloutConfiguration:
+ enableProgressiveRollout: true
  breakingChanges:
  1.0.0:
  message: "This upgrade migrates the SFTP Bulk source to the Airbyte file-based CDK. This is the first necessary step of transitioning a file connector from community to Airbyte maintained."