class_docstr: Helper class that interacts with the SageMaker training service.
class_docstr_tok: ["Helper","class","that","interacts","with","the","SageMaker","training","service","."]
docstr: Build the artifacts and generate the training job request.
docstr_tok: ["Build","the","artifacts","and","generate","the","training","job","request","."]
code: def compile(
      job_settings: _JobSettings,
      job_name: str,
      s3_base_uri: str,
      func: callable,
      func_args: tuple,
      func_kwargs: dict,
      run_info=None,
      serialized_data: _SerializedData = None,
  ) -> dict:
      """Build the artifacts and generate the training job request."""
      from sagemaker.workflow.properties import Properties
      from sagemaker.workflow.parameters import Parameter
      from sagemaker.workflow.execution_variables import (
          ExecutionVariables,
          ExecutionVariable,
      )
      from sagemaker.workflow.utilities import (
          load_step_compilation_context,
      )
  
      step_compilation_context = load_step_compilation_context()
  
      jobs_container_entrypoint = JOBS_CONTAINER_ENTRYPOINT[:]
  
      # generate hmac key for integrity check
      if step_compilation_context is None:
          hmac_key = secrets.token_hex(32)
      else:
          hmac_key = step_compilation_context.function_step_secret_token
  
      # serialize function and arguments
      if step_compilation_context is None:
          stored_function = StoredFunction(
              sagemaker_session=job_settings.sagemaker_session,
              s3_base_uri=s3_base_uri,
              hmac_key=hmac_key,
              s3_kms_key=job_settings.s3_kms_key,
          )
          stored_function.save(func, *func_args, **func_kwargs)
      else:
          stored_function = StoredFunction(
              sagemaker_session=job_settings.sagemaker_session,
              s3_base_uri=s3_base_uri,
              hmac_key=hmac_key,
              s3_kms_key=job_settings.s3_kms_key,
              context=Context(
                  step_name=step_compilation_context.step_name,
                  func_step_s3_dir=step_compilation_context.pipeline_build_time,
              ),
          )
  
          stored_function.save_pipeline_step_function(serialized_data)
  
      stopping_condition = {
          "MaxRuntimeInSeconds": job_settings.max_runtime_in_seconds,
      }
      if job_settings.max_wait_time_in_seconds is not None:
          stopping_condition[
              "MaxWaitTimeInSeconds"
          ] = job_settings.max_wait_time_in_seconds
  
      request_dict = dict(
          TrainingJobName=job_name,
          RoleArn=job_settings.role,
          StoppingCondition=stopping_condition,
          RetryStrategy={
              "MaximumRetryAttempts": job_settings.max_retry_attempts
          },
      )
  
      _update_job_request_with_checkpoint_config(
          func_args, func_kwargs, request_dict
      )
  
      if job_settings.tags:
          request_dict["Tags"] = job_settings.tags
  
      # generate other build artifacts including workspace, requirements.txt
      request_dict["InputDataConfig"] = _generate_input_data_config(
          job_settings=job_settings, s3_base_uri=s3_base_uri
      )
  
      output_config = {"S3OutputPath": s3_base_uri}
      if job_settings.s3_kms_key is not None:
          output_config["KmsKeyId"] = job_settings.s3_kms_key
      request_dict["OutputDataConfig"] = output_config
  
      container_args = ["--s3_base_uri", s3_base_uri]
      container_args.extend(
          ["--region", job_settings.sagemaker_session.boto_region_name]
      )
      container_args.extend(
          [
              "--client_python_version",
              RuntimeEnvironmentManager()._current_python_version(),
          ]
      )
      container_args.extend(
          [
              "--dependency_settings",
              _DependencySettings.from_dependency_file_path(
                  job_settings.dependencies
              ).to_string(),
          ]
      )
      if job_settings.s3_kms_key:
          container_args.extend(
              ["--s3_kms_key", job_settings.s3_kms_key]
          )
  
      if job_settings.job_conda_env:
          container_args.extend(
              ["--job_conda_env", job_settings.job_conda_env]
          )
  
      if step_compilation_context:
          # TODO: remove the duplicates in the list
          container_args.extend(
              [
                  "--pipeline_step_name",
                  step_compilation_context.step_name,
              ]
          )
          container_args.extend(
              [
                  "--pipeline_execution_id",
                  ExecutionVariables.PIPELINE_EXECUTION_ID,
              ]
          )
          container_args.extend(
              [
                  "--func_step_s3_dir",
                  step_compilation_context.pipeline_build_time,
              ]
          )
          container_args.extend(["--property_references"])
          container_args.extend(
              [
                  ExecutionVariables.PIPELINE_EXECUTION_ID.expr["Get"],
                  ExecutionVariables.PIPELINE_EXECUTION_ID.to_string(),
              ]
          )
          for arg in func_args + tuple(func_kwargs.values()):
              if isinstance(
                  arg, (Parameter, ExecutionVariable, Properties)
              ):
                  container_args.extend(
                      [arg.expr["Get"], arg.to_string()]
                  )
  
      if run_info is not None:
          container_args.extend(
              [
                  "--run_in_context",
                  json.dumps(dataclasses.asdict(run_info)),
              ]
          )
      elif _RunContext.get_current_run() is not None:
          container_args.extend(
              [
                  "--run_in_context",
                  _convert_run_to_json(_RunContext.get_current_run()),
              ]
          )
  
      algorithm_spec = dict(
          TrainingImage=job_settings.image_uri,
          TrainingInputMode="File",
          ContainerEntrypoint=jobs_container_entrypoint,
          ContainerArguments=container_args,
      )
  
      request_dict["AlgorithmSpecification"] = algorithm_spec
  
      resource_config = dict(
          VolumeSizeInGB=job_settings.volume_size,
          InstanceCount=job_settings.instance_count,
          InstanceType=job_settings.instance_type,
      )
      if job_settings.volume_kms_key is not None:
          resource_config[
              "VolumeKmsKeyId"
          ] = job_settings.volume_kms_key
      if job_settings.keep_alive_period_in_seconds is not None:
          resource_config[
              "KeepAlivePeriodInSeconds"
          ] = job_settings.keep_alive_period_in_seconds
  
      request_dict["ResourceConfig"] = resource_config
  
      if job_settings.enable_network_isolation is not None:
          request_dict[
              "EnableNetworkIsolation"
          ] = job_settings.enable_network_isolation
  
      if job_settings.encrypt_inter_container_traffic is not None:
          request_dict[
              "EnableInterContainerTrafficEncryption"
          ] = job_settings.encrypt_inter_container_traffic
  
      if job_settings.vpc_config:
          request_dict["VpcConfig"] = job_settings.vpc_config
  
      request_dict[
          "EnableManagedSpotTraining"
      ] = job_settings.use_spot_instances
  
      request_dict["Environment"] = job_settings.environment_variables
      request_dict["Environment"].update(
          {"REMOTE_FUNCTION_SECRET_KEY": hmac_key}
      )
  
      extended_request = _extend_spark_config_to_request(
          request_dict, job_settings, s3_base_uri
      )
  
      return extended_request
  
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 the _Job obj's underlying member method code with example usage: obj.compile(job_settings, job_name, s3_base_uri, func, func_args, func_kwargs, run_info, serialized_data) and returns: extended_request