class_docstr: Object representing a single Python job to execute.
  
  Examples
  --------
  
  Create a new Python job that multiplies two numbers and then adds 5 to the result:
  
  .. code-block:: python
  
      # Create a batch object with a default Python image
  
      b = Batch(default_python_image='hailgenetics/python-dill:3.9-slim')
  
      def multiply(x, y):
          return x * y
  
      def add(x, y):
          return x + y
  
      j = b.new_python_job()
      result = j.call(multiply, 2, 3)
      result = j.call(add, result, 5)
  
      # Write out the str representation of result to a file
  
      b.write_output(result.as_str(), 'hello.txt')
  
      b.run()
  
  Notes
  -----
  This class should never be created directly by the user. Use :meth:`.Batch.new_python_job`
  instead.
class_docstr_tok: ["Object","representing","a","single","Python","job","to","execute",".","Examples","--","--","--","--","Create","a","new","Python","job","that","multiplies","two","numbers","and","then","adds","5","to","the","result",":","..","code-block",":",":","python","#","Create","a","batch","object","with","a","default","Python","image","b","=","Batch","(","default_python_image='hailgenetics\/python-dill:3.9-slim","'",")","def","multiply","(","x",",","y",")",":","return","x","*","y","def","add","(","x",",","y",")",":","return","x","+","y","j","=","b.new_python_job","(",")","result","=","j.call","(","multiply",",","2",",","3",")","result","=","j.call","(","add",",","result",",","5",")","#","Write","out","the","str","representation","of","result","to","a","file","b.write_output","(","result.as_str","(",")",",","'hello.txt","'",")","b.run","(",")","Notes","--","--","-","This","class","should","never","be","created","directly","by","the","user",".","Use",":","meth",":","`",".Batch.new_python_job","`","instead","."]
docstr: Execute a Python function.
  
  Examples
  --------
  
  .. code-block:: python
  
      import json
  
      def add(x, y):
          return x + y
  
      def multiply(x, y):
          return x * y
  
      def format_as_csv(x, y, add_result, mult_result):
          return f'{x},{y},{add_result},{mult_result}'
  
      def csv_to_json(path):
          data = []
          with open(path) as f:
              for line in f:
                  line = line.rstrip()
                  fields = line.split(',')
                  d = {'x': int(fields[0]),
                       'y': int(fields[1]),
                       'add': int(fields[2]),
                       'mult': int(fields[3])}
                  data.append(d)
          return json.dumps(data)
  
  
      # Get all the multiplication and addition table results
  
      b = Batch(name='add-mult-table')
  
      formatted_results = []
  
      for x in range(3):
          for y in range(3):
              j = b.new_python_job(name=f'{x}-{y}')
              add_result = j.call(add, x, y)
              mult_result = j.call(multiply, x, y)
              result = j.call(format_as_csv, x, y, add_result, mult_result)
              formatted_results.append(result.as_str())
  
      cat_j = b.new_bash_job(name='concatenate')
      cat_j.command(f'cat {" ".join(formatted_results)} > {cat_j.output}')
  
      csv_to_json_j = b.new_python_job(name='csv-to-json')
      json_output = csv_to_json_j.call(csv_to_json, cat_j.output)
  
      b.write_output(j.as_str(), '/output/add_mult_table.json')
      b.run()
  
  Notes
  -----
  Unlike the :class:`.BashJob`, a :class:`.PythonJob` returns a new
  :class:`.PythonResult` for every invocation of :meth:`.PythonJob.call`. A
  :class:`.PythonResult` can be used as an argument in subsequent invocations of
  :meth:`.PythonJob.call`, as an argument in downstream python jobs,
  or as inputs to other bash jobs. Likewise, :class:`.InputResourceFile`,
  :class:`.JobResourceFile`, and :class:`.ResourceGroup` can be passed to
  :meth:`.PythonJob.call`. Batch automatically detects dependencies between jobs
  including between python jobs and bash jobs.
  
  When a :class:`.ResourceFile` is passed as an argument, it is passed to the
  function as a string to the local file path. When a :class:`.ResourceGroup`
  is passed as an argument, it is passed to the function as a dict where the
  keys are the resource identifiers in the original :class:`.ResourceGroup`
  and the values are the local file paths.
  
  Like :class:`.JobResourceFile`, all :class:`.PythonResult` are stored as
  temporary files and must be written to a permanent location using
  :meth:`.Batch.write_output` if the output needs to be saved. A
  PythonResult is saved as a dill serialized object. However, you
  can use one of the methods :meth:`.PythonResult.as_str`, :meth:`.PythonResult.as_repr`,
  or :meth:`.PythonResult.as_json` to convert a `PythonResult` to a
  `JobResourceFile` with the desired output.
  
  Warning
  -------
  
  You must have any non-builtin packages that are used by `unapplied` installed
  in your image. You can use :func:`.docker.build_python_image` to build a
  Python image with additional Python packages installed that is compatible
  with Python jobs.
  
  Here are some tips to make sure your function can be used with Batch:
  
   - Only reference top-level modules in your functions: like numpy or pandas.
   - If you get a serialization error, try moving your imports into your function.
   - Instead of serializing a complex class, determine what information is essential
     and only serialize that, perhaps as a dict or array.
  
  Parameters
  ----------
  unapplied:
      A reference to a Python function to execute.
  args:
      Positional arguments to the Python function. Must be either a builtin
      Python object, a :class:`.Resource`, or a Dill serializable object.
  kwargs:
      Key-word arguments to the Python function. Must be either a builtin
      Python object, a :class:`.Resource`, or a Dill serializable object.
  
  Returns
  -------
  :class:`.resource.PythonResult`
docstr_tok: ["Execute","a","Python","function",".","Examples","--","--","--","--","..","code-block",":",":","python","import","json","def","add","(","x",",","y",")",":","return","x","+","y","def","multiply","(","x",",","y",")",":","return","x","*","y","def","format_as_csv","(","x",",","y",",","add_result",",","mult_result",")",":","return","f","'","{","x","}",",","{","y","}",",","{","add_result","}",",","{","mult_result","}","'","def","csv_to_json","(","path",")",":","data","=","[","]","with","open","(","path",")","as","f",":","for","line","in","f",":","line","=","line.rstrip","(",")","fields","=","line.split","(","'",",","'",")","d","=","{","'","x","'",":","int","(","fields","[","0","]",")",",","'","y","'",":","int","(","fields","[","1","]",")",",","'add","'",":","int","(","fields","[","2","]",")",",","'mult","'",":","int","(","fields","[","3","]",")","}","data.append","(","d",")","return","json.dumps","(","data",")","#","Get","all","the","multiplication","and","addition","table","results","b","=","Batch","(","name='add-mult-table","'",")","formatted_results","=","[","]","for","x","in","range","(","3",")",":","for","y","in","range","(","3",")",":","j","=","b.new_python_job","(","name=f","'","{","x","}","-","{","y","}","'",")","add_result","=","j.call","(","add",",","x",",","y",")","mult_result","=","j.call","(","multiply",",","x",",","y",")","result","=","j.call","(","format_as_csv",",","x",",","y",",","add_result",",","mult_result",")","formatted_results.append","(","result.as_str","(",")",")","cat_j","=","b.new_bash_job","(","name='concatenate","'",")","cat_j.command","(","f'cat","{","``","``",".join","(","formatted_results",")","}",">","{","cat_j.output","}","'",")","csv_to_json_j","=","b.new_python_job","(","name='csv-to-json","'",")","json_output","=","csv_to_json_j.call","(","csv_to_json",",","cat_j.output",")","b.write_output","(","j.as_str","(",")",",","'\/output\/add_mult_table.json","'",")","b.run","(",")","Notes","--","--","-","Unlike","the",":","class",":","`",".BashJob","`",",","a",":","class",":","`",".PythonJob","`","returns","a","new",":","class",":","`",".PythonResult","`","for","every","invocation","of",":","meth",":","`",".PythonJob.call","`",".","A",":","class",":","`",".PythonResult","`","can","be","used","as","an","argument","in","subsequent","invocations","of",":","meth",":","`",".PythonJob.call","`",",","as","an","argument","in","downstream","python","jobs",",","or","as","inputs","to","other","bash","jobs",".","Likewise",",",":","class",":","`",".InputResourceFile","`",",",":","class",":","`",".JobResourceFile","`",",","and",":","class",":","`",".ResourceGroup","`","can","be","passed","to",":","meth",":","`",".PythonJob.call","`",".","Batch","automatically","detects","dependencies","between","jobs","including","between","python","jobs","and","bash","jobs",".","When","a",":","class",":","`",".ResourceFile","`","is","passed","as","an","argument",",","it","is","passed","to","the","function","as","a","string","to","the","local","file","path",".","When","a",":","class",":","`",".ResourceGroup","`","is","passed","as","an","argument",",","it","is","passed","to","the","function","as","a","dict","where","the","keys","are","the","resource","identifiers","in","the","original",":","class",":","`",".ResourceGroup","`","and","the","values","are","the","local","file","paths",".","Like",":","class",":","`",".JobResourceFile","`",",","all",":","class",":","`",".PythonResult","`","are","stored","as","temporary","files","and","must","be","written","to","a","permanent","location","using",":","meth",":","`",".Batch.write_output","`","if","the","output","needs","to","be","saved",".","A","PythonResult","is","saved","as","a","dill","serialized","object",".","However",",","you","can","use","one","of","the","methods",":","meth",":","`",".PythonResult.as_str","`",",",":","meth",":","`",".PythonResult.as_repr","`",",","or",":","meth",":","`",".PythonResult.as_json","`","to","convert","a","`","PythonResult","`","to","a","`","JobResourceFile","`","with","the","desired","output",".","Warning","--","--","--","-","You","must","have","any","non-builtin","packages","that","are","used","by","`","unapplied","`","installed","in","your","image",".","You","can","use",":","func",":","`",".docker.build_python_image","`","to","build","a","Python","image","with","additional","Python","packages","installed","that","is","compatible","with","Python","jobs",".","Here","are","some","tips","to","make","sure","your","function","can","be","used","with","Batch",":","-","Only","reference","top-level","modules","in","your","functions",":","like","numpy","or","pandas",".","-","If","you","get","a","serialization","error",",","try","moving","your","imports","into","your","function",".","-","Instead","of","serializing","a","complex","class",",","determine","what","information","is","essential","and","only","serialize","that",",","perhaps","as","a","dict","or","array",".","Parameters","--","--","--","--","--","unapplied",":","A","reference","to","a","Python","function","to","execute",".","args",":","Positional","arguments","to","the","Python","function",".","Must","be","either","a","builtin","Python","object",",","a",":","class",":","`",".Resource","`",",","or","a","Dill","serializable","object",".","kwargs",":","Key-word","arguments","to","the","Python","function",".","Must","be","either","a","builtin","Python","object",",","a",":","class",":","`",".Resource","`",",","or","a","Dill","serializable","object",".","Returns","--","--","--","-",":","class",":","`",".resource.PythonResult","`"]
code: def call(
      self,
      unapplied: Callable,
      *args: UnpreparedArg,
      **kwargs: UnpreparedArg,
  ) -> "_resource.PythonResult":
      """Execute a Python function.
  
      Examples
      --------
  
      .. code-block:: python
  
          import json
  
          def add(x, y):
              return x + y
  
          def multiply(x, y):
              return x * y
  
          def format_as_csv(x, y, add_result, mult_result):
              return f'{x},{y},{add_result},{mult_result}'
  
          def csv_to_json(path):
              data = []
              with open(path) as f:
                  for line in f:
                      line = line.rstrip()
                      fields = line.split(',')
                      d = {'x': int(fields[0]),
                           'y': int(fields[1]),
                           'add': int(fields[2]),
                           'mult': int(fields[3])}
                      data.append(d)
              return json.dumps(data)
  
  
          # Get all the multiplication and addition table results
  
          b = Batch(name='add-mult-table')
  
          formatted_results = []
  
          for x in range(3):
              for y in range(3):
                  j = b.new_python_job(name=f'{x}-{y}')
                  add_result = j.call(add, x, y)
                  mult_result = j.call(multiply, x, y)
                  result = j.call(format_as_csv, x, y, add_result, mult_result)
                  formatted_results.append(result.as_str())
  
          cat_j = b.new_bash_job(name='concatenate')
          cat_j.command(f'cat {" ".join(formatted_results)} > {cat_j.output}')
  
          csv_to_json_j = b.new_python_job(name='csv-to-json')
          json_output = csv_to_json_j.call(csv_to_json, cat_j.output)
  
          b.write_output(j.as_str(), '/output/add_mult_table.json')
          b.run()
  
      Notes
      -----
      Unlike the :class:`.BashJob`, a :class:`.PythonJob` returns a new
      :class:`.PythonResult` for every invocation of :meth:`.PythonJob.call`. A
      :class:`.PythonResult` can be used as an argument in subsequent invocations of
      :meth:`.PythonJob.call`, as an argument in downstream python jobs,
      or as inputs to other bash jobs. Likewise, :class:`.InputResourceFile`,
      :class:`.JobResourceFile`, and :class:`.ResourceGroup` can be passed to
      :meth:`.PythonJob.call`. Batch automatically detects dependencies between jobs
      including between python jobs and bash jobs.
  
      When a :class:`.ResourceFile` is passed as an argument, it is passed to the
      function as a string to the local file path. When a :class:`.ResourceGroup`
      is passed as an argument, it is passed to the function as a dict where the
      keys are the resource identifiers in the original :class:`.ResourceGroup`
      and the values are the local file paths.
  
      Like :class:`.JobResourceFile`, all :class:`.PythonResult` are stored as
      temporary files and must be written to a permanent location using
      :meth:`.Batch.write_output` if the output needs to be saved. A
      PythonResult is saved as a dill serialized object. However, you
      can use one of the methods :meth:`.PythonResult.as_str`, :meth:`.PythonResult.as_repr`,
      or :meth:`.PythonResult.as_json` to convert a `PythonResult` to a
      `JobResourceFile` with the desired output.
  
      Warning
      -------
  
      You must have any non-builtin packages that are used by `unapplied` installed
      in your image. You can use :func:`.docker.build_python_image` to build a
      Python image with additional Python packages installed that is compatible
      with Python jobs.
  
      Here are some tips to make sure your function can be used with Batch:
  
       - Only reference top-level modules in your functions: like numpy or pandas.
       - If you get a serialization error, try moving your imports into your function.
       - Instead of serializing a complex class, determine what information is essential
         and only serialize that, perhaps as a dict or array.
  
      Parameters
      ----------
      unapplied:
          A reference to a Python function to execute.
      args:
          Positional arguments to the Python function. Must be either a builtin
          Python object, a :class:`.Resource`, or a Dill serializable object.
      kwargs:
          Key-word arguments to the Python function. Must be either a builtin
          Python object, a :class:`.Resource`, or a Dill serializable object.
  
      Returns
      -------
      :class:`.resource.PythonResult`
      """
  
      if not callable(unapplied):
          raise BatchException(
              f"unapplied must be a callable function. Found {type(unapplied)}."
          )
  
      if asyncio.iscoroutinefunction(unapplied):
          unapplied_copy = unapplied
  
          def run_async(*args, **kwargs):
              return asyncio.run(unapplied_copy(*args, **kwargs))
  
          unapplied = run_async
  
      for arg in args:
          if isinstance(arg, Job):
              raise BatchException(
                  "arguments to a PythonJob cannot be other job objects."
              )
  
      for value in kwargs.values():
          if isinstance(value, Job):
              raise BatchException(
                  "arguments to a PythonJob cannot be other job objects."
              )
  
      # Some builtins like `print` do not have signatures
      try:
          inspect.signature(unapplied).bind(*args, **kwargs)
      except ValueError as e:
          # Some builtins like `print` don't have a signature that inspect can read
          if "no signature found for builtin" not in e.args[0]:
              raise e
      except TypeError as e:
          raise BatchException(
              f"Cannot call {unapplied.__name__} with the supplied arguments"
          ) from e
  
      def handle_arg(r):
          if r._source != self:
              self._add_inputs(r)
              if r._source is not None:
                  if r not in r._source._valid:
                      name = r._source._resources_inverse[r]
                      raise BatchException(
                          f"undefined resource '{name}'\n"
                      )
                  self._dependencies.add(r._source)
                  r._source._add_internal_outputs(r)
          else:
              _add_resource_to_set(self._valid, r)
  
          self._mentioned.add(r)
  
      def handle_args(r):
          if isinstance(r, _resource.Resource):
              handle_arg(r)
          elif isinstance(r, (list, tuple)):
              for elt in r:
                  handle_args(elt)
          elif isinstance(r, dict):
              for v in r.values():
                  handle_args(v)
  
      handle_args(args)
      handle_args(kwargs)
  
      self.n_results += 1
      result = self._get_python_resource(f"result{self.n_results}")
      handle_arg(result)
  
      unapplied_id = self._batch._register_python_function(unapplied)
  
      self._function_calls.append((result, unapplied_id, args, kwargs))
  
      return result
  
imports: import asyncio
  import inspect
  import os
  import re
  import textwrap
  import warnings
  from shlex import quote
  from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union, cast, Literal
  from typing_extensions import Self
  import hailtop.batch_client.client
  from .None import backend, batch
  from .None import resource
  from .exceptions import BatchException
label_desc: Use image node_id 4 for calling the PythonJob obj's underlying member method code with example usage: obj.call(unapplied) and returns: result, asyncio