Review

Author: Ilyas Gasanov

.env.docker

@@ -113,7 +113,7 @@ TEST_SFTP_PORT_FOR_CONFTEST=2222
 TEST_SFTP_HOST_FOR_WORKER=test-sftp
 TEST_SFTP_PORT_FOR_WORKER=2222
 TEST_SFTP_USER=syncmaster
-TEST_SFTP_PASSWORD=AesujeifohgoaCu0Boosiet5aimeitho
+TEST_SFTP_PASSWORD=test_only
 
 SPARK_CONF_DIR=/app/tests/spark/hive/conf/
 HADOOP_CONF_DIR=/app/tests/spark/hadoop/

.env.local

@@ -100,7 +100,7 @@ export TEST_SFTP_PORT_FOR_CONFTEST=2222
 export TEST_SFTP_HOST_FOR_WORKER=test-sftp
 export TEST_SFTP_PORT_FOR_WORKER=2222
 export TEST_SFTP_USER=syncmaster
-export TEST_SFTP_PASSWORD=AesujeifohgoaCu0Boosiet5aimeitho
+export TEST_SFTP_PASSWORD=test_only
 
 export SPARK_CONF_DIR=./tests/spark/hive/conf/
 export HADOOP_CONF_DIR=./tests/spark/hadoop/

docker-compose.test.yml

@@ -277,7 +277,7 @@ services:
       USER_NAME: syncmaster
       PASSWORD_ACCESS: true
       SUDO_ACCESS: true
-      USER_PASSWORD: AesujeifohgoaCu0Boosiet5aimeitho
+      USER_PASSWORD: test_only
     profiles: [sftp, all]
 
 volumes:

syncmaster/dto/transfers.py

@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 import json
 from dataclasses import dataclass
+from pathlib import Path
 from typing import ClassVar
 
 from onetl.file.format import CSV, JSON, ORC, XML, Excel, JSONLine, Parquet
@@ -98,3 +99,4 @@ class HDFSTransferDTO(FileTransferDTO):
 @dataclass
 class SFTPTransferDTO(FileTransferDTO):
     type: ClassVar[str] = "sftp"
+    temp_worker_directory_path: Path | None = None

syncmaster/worker/controller.py

@@ -1,6 +1,7 @@
 # SPDX-FileCopyrightText: 2023-2024 MTS PJSC
 # SPDX-License-Identifier: Apache-2.0
 import logging
+import tempfile
 from typing import Any
 
 from syncmaster.db.models import Connection, Run
@@ -128,8 +129,21 @@ def perform_transfer(self, settings: WorkerAppSettings) -> None:
             self.source_handler.connect(spark)
             self.target_handler.connect(spark)
 
-            df = self.source_handler.read()
-            self.target_handler.write(df)
+            source_needs_temp_dir = self.source_handler.transfer_dto.type == "sftp"
+            target_needs_temp_dir = self.target_handler.transfer_dto.type == "sftp"
+
+            if source_needs_temp_dir or target_needs_temp_dir:
+                with tempfile.TemporaryDirectory(prefix="syncmaster_") as temp_dir:
+                    if source_needs_temp_dir:
+                        self.source_handler.transfer_dto.temp_worker_directory_path = temp_dir
+                    if target_needs_temp_dir:
+                        self.target_handler.transfer_dto.temp_worker_directory_path = temp_dir
+
+                    df = self.source_handler.read()
+                    self.target_handler.write(df)
+            else:
+                df = self.source_handler.read()
+                self.target_handler.write(df)
 
     def get_handler(
         self,

syncmaster/worker/handlers/file/sftp.py

@@ -24,7 +24,7 @@ def connect(self, spark: SparkSession) -> None:
             port=self.connection_dto.port,
             user=self.connection_dto.user,
             password=self.connection_dto.password,
-            compress=False,
+            compress=False,  # to avoid errors from combining file and SCP-level compression
         ).check()
         self.local_connection = SparkLocalFS(
             spark=spark,
@@ -36,16 +36,14 @@ def read(self) -> DataFrame:
         downloader = FileDownloader(
             connection=self.connection,
             source_path=self.transfer_dto.directory_path,
-            temp_path="/tmp/syncmaster",
-            local_path="/tmp/syncmaster/sftp",
-            options={"if_exists": "replace_entire_directory"},
+            local_path=self.transfer_dto.temp_worker_directory_path,
         )
         downloader.run()
 
         reader = FileDFReader(
             connection=self.local_connection,
             format=self.transfer_dto.file_format,
-            source_path="/tmp/syncmaster/sftp",
+            source_path=self.transfer_dto.temp_worker_directory_path,
             df_schema=StructType.fromJson(self.transfer_dto.df_schema) if self.transfer_dto.df_schema else None,
             options=self.transfer_dto.options,
         )
@@ -65,16 +63,14 @@ def write(self, df: DataFrame) -> None:
         writer = FileDFWriter(
             connection=self.local_connection,
             format=self.transfer_dto.file_format,
-            target_path="/tmp/syncmaster/sftp",
+            target_path=self.transfer_dto.temp_worker_directory_path,
             options=self.transfer_dto.options,
         )
         writer.run(df=df)
 
         uploader = FileUploader(
             connection=self.connection,
-            local_path="/tmp/syncmaster/sftp",
-            temp_path="/config/target",  # SFTP host
+            local_path=self.transfer_dto.temp_worker_directory_path,
             target_path=self.transfer_dto.directory_path,
-            options={"if_exists": "replace_entire_directory"},
         )
         uploader.run()

tests/test_integration/test_run_transfer/connection_fixtures/sftp_fixtures.py

@@ -84,7 +84,7 @@ def sftp_file_connection(sftp_for_conftest):
         port=sftp_for_conftest.port,
         user=sftp_for_conftest.user,
         password=sftp_for_conftest.password,
-        compress=False,
+        compress=False,  # to avoid errors from combining file and SCP-level compression
     )
 
 

tests/test_integration/test_run_transfer/test_sftp.py

@@ -1,5 +1,6 @@
 import os
 import secrets
+from pathlib import Path
 
 import pytest
 import pytest_asyncio
@@ -246,6 +247,7 @@ async def test_run_transfer_postgres_to_sftp(
     postgres_to_sftp: Transfer,
     target_file_format,
     file_format_flavor: str,
+    tmp_path: Path,
 ):
     format_name, format = target_file_format
 
@@ -279,18 +281,15 @@ async def test_run_transfer_postgres_to_sftp(
     downloader = FileDownloader(
         connection=sftp_file_connection,
         source_path=f"/config/target/{format_name}/{file_format_flavor}",
-        temp_path="/tmp/syncmaster",
-        local_path="/tmp/syncmaster/sftp",
-        options={"if_exists": "replace_entire_directory"},
+        local_path=tmp_path,
     )
     downloader.run()
 
     reader = FileDFReader(
         connection=sftp_file_df_connection,
         format=format,
-        source_path="/tmp/syncmaster/sftp",
+        source_path=tmp_path,
         df_schema=init_df.schema,
-        options={},
     )
     df = reader.run()