Move path splitting to helper method

src/snowflake/snowpark/_internal/utils.py

@@ -284,6 +284,12 @@ def normalize_local_file(file: str) -> str:
     return normalize_path(file, is_local=True)
 
 
+def split_path(path: str) -> Tuple[str, str]:
+    """Split a file path into directory and file name."""
+    path = unwrap_single_quote(path)
+    return path.rsplit("/", maxsplit=1)
+
+
 def unwrap_stage_location_single_quote(name: str) -> str:
     new_name = unwrap_single_quote(name)
     if any(new_name.startswith(prefix) for prefix in SNOWFLAKE_PATH_PREFIXES):

src/snowflake/snowpark/file_operation.py

@@ -18,7 +18,7 @@
     normalize_local_file,
     normalize_remote_file_or_dir,
     result_set_to_rows,
-    unwrap_single_quote,
+    split_path,
 )
 
 
@@ -274,9 +274,7 @@ def put_stream(
                 )
                 raise ne.with_traceback(tb) from None
         else:
-            stage_with_prefix, dest_filename = unwrap_single_quote(
-                stage_location
-            ).rsplit("/", maxsplit=1)
+            stage_with_prefix, dest_filename = split_path(stage_location)
             put_result = self._session._conn.upload_stream(
                 input_stream=input_stream,
                 stage_location=stage_with_prefix,
@@ -341,9 +339,7 @@ def get_stream(
         else:
             options = {"parallel": parallel}
             tmp_dir = tempfile.gettempdir()
-            src_file_name = unwrap_single_quote(stage_location).rsplit("/", maxsplit=1)[
-                1
-            ]
+            src_file_name = split_path(stage_location)[1]
             local_file_name = os.path.join(tmp_dir, src_file_name)
             plan = self._session._plan_builder.file_operation_plan(
                 "get",