code: def create_job(
      self,
      step_name,
      step_cli,
      task_spec,
      code_package_sha,
      code_package_url,
      code_package_ds,
      image,
      queue,
      iam_role=None,
      execution_role=None,
      cpu=None,
      gpu=None,
      memory=None,
      run_time_limit=None,
      shared_memory=None,
      max_swap=None,
      swappiness=None,
      inferentia=None,
      efa=None,
      env={},
      attrs={},
      host_volumes=None,
      efs_volumes=None,
      use_tmpfs=None,
      tmpfs_tempdir=None,
      tmpfs_size=None,
      tmpfs_path=None,
      num_parallel=0,
  ):
      job_name = self._job_name(
          attrs.get("metaflow.user"),
          attrs.get("metaflow.flow_name"),
          attrs.get("metaflow.run_id"),
          attrs.get("metaflow.step_name"),
          attrs.get("metaflow.task_id"),
          attrs.get("metaflow.retry_count"),
      )
      job = (
          self._client.job()
          .job_name(job_name)
          .job_queue(queue)
          .command(
              self._command(
                  self.environment,
                  code_package_url,
                  step_name,
                  [step_cli],
                  task_spec,
              )
          )
          .image(image)
          .iam_role(iam_role)
          .execution_role(execution_role)
          .cpu(cpu)
          .gpu(gpu)
          .memory(memory)
          .shared_memory(shared_memory)
          .max_swap(max_swap)
          .swappiness(swappiness)
          .inferentia(inferentia)
          .efa(efa)
          .timeout_in_secs(run_time_limit)
          .job_def(
              image,
              iam_role,
              queue,
              execution_role,
              shared_memory,
              max_swap,
              swappiness,
              inferentia,
              efa,
              memory=memory,
              host_volumes=host_volumes,
              efs_volumes=efs_volumes,
              use_tmpfs=use_tmpfs,
              tmpfs_tempdir=tmpfs_tempdir,
              tmpfs_size=tmpfs_size,
              tmpfs_path=tmpfs_path,
              num_parallel=num_parallel,
          )
          .task_id(attrs.get("metaflow.task_id"))
          .environment_variable(
              "AWS_DEFAULT_REGION", self._client.region()
          )
          .environment_variable("METAFLOW_CODE_SHA", code_package_sha)
          .environment_variable("METAFLOW_CODE_URL", code_package_url)
          .environment_variable("METAFLOW_CODE_DS", code_package_ds)
          .environment_variable("METAFLOW_USER", attrs["metaflow.user"])
          .environment_variable(
              "METAFLOW_SERVICE_URL", SERVICE_INTERNAL_URL
          )
          .environment_variable(
              "METAFLOW_SERVICE_HEADERS", json.dumps(SERVICE_HEADERS)
          )
          .environment_variable(
              "METAFLOW_DATASTORE_SYSROOT_S3", DATASTORE_SYSROOT_S3
          )
          .environment_variable(
              "METAFLOW_DATATOOLS_S3ROOT", DATATOOLS_S3ROOT
          )
          .environment_variable("METAFLOW_DEFAULT_DATASTORE", "s3")
          .environment_variable(
              "METAFLOW_DEFAULT_METADATA", DEFAULT_METADATA
          )
          .environment_variable("METAFLOW_CARD_S3ROOT", CARD_S3ROOT)
          .environment_variable(
              "METAFLOW_RUNTIME_ENVIRONMENT", "aws-batch"
          )
      )
  
      # Temporary passing of *some* environment variables. Do not rely on this
      # mechanism as it will be removed in the near future
      for k, v in config_values():
          if k.startswith("METAFLOW_CONDA_") or k.startswith(
              "METAFLOW_DEBUG_"
          ):
              job.environment_variable(k, v)
  
      if DEFAULT_SECRETS_BACKEND_TYPE is not None:
          job.environment_variable(
              "METAFLOW_DEFAULT_SECRETS_BACKEND_TYPE",
              DEFAULT_SECRETS_BACKEND_TYPE,
          )
      if AWS_SECRETS_MANAGER_DEFAULT_REGION is not None:
          job.environment_variable(
              "METAFLOW_AWS_SECRETS_MANAGER_DEFAULT_REGION",
              AWS_SECRETS_MANAGER_DEFAULT_REGION,
          )
  
      tmpfs_enabled = use_tmpfs or (tmpfs_size and not use_tmpfs)
  
      if tmpfs_enabled and tmpfs_tempdir:
          job.environment_variable("METAFLOW_TEMPDIR", tmpfs_path)
  
      if S3_SERVER_SIDE_ENCRYPTION is not None:
          job.environment_variable(
              "METAFLOW_S3_SERVER_SIDE_ENCRYPTION",
              S3_SERVER_SIDE_ENCRYPTION,
          )
  
      # Skip setting METAFLOW_DATASTORE_SYSROOT_LOCAL because metadata sync between the local user
      # instance and the remote AWS Batch instance assumes metadata is stored in DATASTORE_LOCAL_DIR
      # on the remote AWS Batch instance; this happens when METAFLOW_DATASTORE_SYSROOT_LOCAL
      # is NOT set (see get_datastore_root_from_config in datastore/local.py).
      # add METAFLOW_S3_ENDPOINT_URL
      if S3_ENDPOINT_URL is not None:
          job.environment_variable(
              "METAFLOW_S3_ENDPOINT_URL", S3_ENDPOINT_URL
          )
  
      for name, value in env.items():
          job.environment_variable(name, value)
  
      if attrs:
          for key, value in attrs.items():
              job.parameter(key, value)
      # Tags for AWS Batch job (for say cost attribution)
      if BATCH_EMIT_TAGS:
          job.tag("app", "metaflow")
          for key in [
              "metaflow.flow_name",
              "metaflow.run_id",
              "metaflow.step_name",
              "metaflow.run_id.$",
              "metaflow.production_token",
          ]:
              if key in attrs:
                  job.tag(key, attrs.get(key))
          # As some values can be affected by users, sanitize them so they adhere to AWS tagging restrictions.
          for key in [
              "metaflow.version",
              "metaflow.user",
              "metaflow.owner",
          ]:
              if key in attrs:
                  k, v = sanitize_batch_tag(key, attrs.get(key))
                  job.tag(k, v)
      return job
  
imports: import atexit
  import copy
  import json
  import os
  import select
  import shlex
  import time
  from metaflow import util
  from metaflow.plugins.datatools.s3.s3tail import S3Tail
  from metaflow.plugins.aws.aws_utils import sanitize_batch_tag
  from metaflow.exception import MetaflowException
  from metaflow.metaflow_config import SERVICE_INTERNAL_URL, DATATOOLS_S3ROOT, DATASTORE_SYSROOT_S3, DEFAULT_METADATA, SERVICE_HEADERS, BATCH_EMIT_TAGS, CARD_S3ROOT, S3_ENDPOINT_URL, DEFAULT_SECRETS_BACKEND_TYPE, AWS_SECRETS_MANAGER_DEFAULT_REGION, S3_SERVER_SIDE_ENCRYPTION
  from metaflow.metaflow_config_funcs import config_values
  from metaflow.mflog import export_mflog_env_vars, bash_capture_logs, tail_logs, BASH_SAVE_LOGS
  from .batch_client import BatchClient
label_desc: Use image node_id 7 for calling the Batch obj's underlying member method code with example usage: obj.create_job(step_name, step_cli, task_spec, code_package_sha, code_package_url, code_package_ds, image, queue, iam_role, execution_role, cpu, gpu, memory, run_time_limit, shared_memory, max_swap, swappiness, inferentia, efa, env, attrs, host_volumes, efs_volumes, use_tmpfs, tmpfs_tempdir, tmpfs_size, tmpfs_path, num_parallel) and returns: job