code: def _prepare_and_upload_workspace(
      local_dependencies_path: str,
      include_local_workdir: bool,
      pre_execution_commands: List[str],
      pre_execution_script_local_path: str,
      s3_base_uri: str,
      s3_kms_key: str,
      sagemaker_session: Session,
      custom_file_filter: Optional[
          Union[Callable[[str, List], List], CustomFileFilter]
      ] = None,
  ) -> str:
      """Prepare and upload the workspace to S3.
  
      Under pipeline context, only workdir is packaged in the workspace folder and uploaded to s3.
      Under remote function context, workdir along with pre execution scripts and dependencies
        are packaged together into the workspace folder and uploaded to S3.
      """
      from sagemaker.workflow.utilities import (
          load_step_compilation_context,
      )
  
      step_compilation_context = load_step_compilation_context()
  
      if not (
          local_dependencies_path
          or include_local_workdir
          or pre_execution_commands
          or pre_execution_script_local_path
      ):
          return None
  
      func_step_s3_dir = None
      if step_compilation_context:
          func_step_s3_dir = (
              step_compilation_context.pipeline_build_time
          )
          if not include_local_workdir:
              return None
          if not step_compilation_context.upload_workspace:
              return s3_path_join(
                  s3_base_uri,
                  REMOTE_FUNCTION_WORKSPACE,
                  func_step_s3_dir,
              )
  
      with _tmpdir() as tmp_dir:
          tmp_workspace_dir = os.path.join(tmp_dir, "temp_workspace/")
          os.mkdir(tmp_workspace_dir)
          # TODO Remove the following hack to avoid dir_exists error in the copy_tree call below.
          tmp_workspace = os.path.join(
              tmp_workspace_dir, JOB_REMOTE_FUNCTION_WORKSPACE
          )
  
          if include_local_workdir:
              copy_workdir(tmp_workspace, custom_file_filter)
              logger.info(
                  "Copied user workspace to '%s'", tmp_workspace
              )
  
          if not os.path.isdir(tmp_workspace):
              # create the directory if no workdir_path was provided in the input.
              os.mkdir(tmp_workspace)
  
          if not step_compilation_context:
              _prepare_dependencies_and_pre_execution_scripts(
                  local_dependencies_path=local_dependencies_path,
                  pre_execution_commands=pre_execution_commands,
                  pre_execution_script_local_path=pre_execution_script_local_path,
                  s3_base_uri=s3_base_uri,
                  s3_kms_key=s3_kms_key,
                  sagemaker_session=sagemaker_session,
                  tmp_dir=tmp_workspace,
              )
  
          workspace_archive_path = os.path.join(tmp_dir, "workspace")
          workspace_archive_path = shutil.make_archive(
              workspace_archive_path, "zip", tmp_workspace_dir
          )
          logger.info(
              "Successfully created workdir archive at '%s'",
              workspace_archive_path,
          )
  
          upload_path = S3Uploader.upload(
              workspace_archive_path,
              s3_path_join(
                  s3_base_uri,
                  REMOTE_FUNCTION_WORKSPACE,
                  func_step_s3_dir,
              ),
              s3_kms_key,
              sagemaker_session,
          )
          logger.info(
              "Successfully uploaded workdir to '%s'", upload_path
          )
          if step_compilation_context:
              step_compilation_context.upload_workspace = False
          return upload_path
  
imports: from __future__ import absolute_import
  import dataclasses
  import os
  import re
  import shutil
  import sys
  import json
  import secrets
  from typing import Callable, Dict, List, Optional, Tuple, Union, TYPE_CHECKING
  from urllib.parse import urlparse
  from io import BytesIO
  from sagemaker.config.config_schema import REMOTE_FUNCTION_ENVIRONMENT_VARIABLES, REMOTE_FUNCTION_IMAGE_URI, REMOTE_FUNCTION_DEPENDENCIES, REMOTE_FUNCTION_PRE_EXECUTION_COMMANDS, REMOTE_FUNCTION_PRE_EXECUTION_SCRIPT, REMOTE_FUNCTION_INCLUDE_LOCAL_WORKDIR, REMOTE_FUNCTION_INSTANCE_TYPE, REMOTE_FUNCTION_JOB_CONDA_ENV, REMOTE_FUNCTION_ROLE_ARN, REMOTE_FUNCTION_S3_ROOT_URI, REMOTE_FUNCTION_S3_KMS_KEY_ID, REMOTE_FUNCTION_VOLUME_KMS_KEY_ID, REMOTE_FUNCTION_TAGS, REMOTE_FUNCTION_VPC_CONFIG_SUBNETS, REMOTE_FUNCTION_VPC_CONFIG_SECURITY_GROUP_IDS, REMOTE_FUNCTION_ENABLE_INTER_CONTAINER_TRAFFIC_ENCRYPTION
  from sagemaker.experiments._run_context import _RunContext
  from sagemaker.experiments.run import Run
  from sagemaker.image_uris import get_base_python_image_uri
  from sagemaker import image_uris
  from sagemaker.remote_function.checkpoint_location import CheckpointLocation
  from sagemaker.session import get_execution_role, _logs_for_job, Session
  from sagemaker.utils import name_from_base, _tmpdir, resolve_value_from_config, format_tags, Tags
  from sagemaker.s3 import s3_path_join, S3Uploader
  from sagemaker import vpc_utils
  from sagemaker.remote_function.core.stored_function import StoredFunction, _SerializedData
  from sagemaker.remote_function.core.pipeline_variables import Context
  from sagemaker.remote_function.runtime_environment.runtime_environment_manager import RuntimeEnvironmentManager, _DependencySettings
  from sagemaker.remote_function import logging_config
  from sagemaker.remote_function.spark_config import SparkConfig
  from sagemaker.spark import defaults
  from sagemaker.remote_function.custom_file_filter import CustomFileFilter, copy_workdir, resolve_custom_file_filter_from_config_file
label_desc: Use image node_id 4 for calling a global function with example usage: _prepare_and_upload_workspace(local_dependencies_path, include_local_workdir, pre_execution_commands, pre_execution_script_local_path, s3_base_uri, s3_kms_key, sagemaker_session, custom_file_filter) and returns: None, upload_path, None, s3_path_join