def parse_location()

in wstl1/tools/notebook/extensions/wstl/magics/_location.py [0:0]

• 28 lines of code
• 10 McCabe index (conditional complexity)

def parse_location(shell, input_wstl_arg, file_ext=None, load_contents=True):
  r"""Parses the argument and returns a Location protobuf message.

  Input arguments with the following prefixes are supported:
  * json://{\"hello\": \"world\"} - using python dict and list notation to
  define a json object or list.
  * file://<file_path> - path to file on local file system.
  * gs://<gcs_path> - path to file on Google Cloud Storage.
  * py://<name_of_python_variable> - name of python variable instantiated within
  session. Note that a python list will be parsed as a single JSON Array and a
  single Location protobuf message is created. If list entries should be parsed
  separately, use pylist instead.
  * pylist://<name_of_python_variable> - name of python list variable
  instantiated within session. Each entry in the list will be parsed as a
  separate Location protobuf message.

  Args:
    shell: ipython interactive shell.
    input_wstl_arg: wstl magic command input argument.
    file_ext: list of valid file extensions. Either `_constants.JSON_FILE_EXT`
      or `WSTL_FILE_EXT`.
    load_contents: flag indicating whether to load contents from disk instead of
      storing a path.

  Returns:
    A Location protobuf message.

  Raises:
    ValueError: An unknown location prefix or the python variable can not be
    encoded into json.
  """
  (inputs, gcs_paths) = _parser.parse_object(
      shell, input_wstl_arg, file_ext=file_ext, load_contents=load_contents)
  if input_wstl_arg.startswith(_constants.JSON_ARG_PREFIX) or \
      input_wstl_arg.startswith(_constants.PYTHON_ARG_PREFIX) or \
      input_wstl_arg.startswith(_constants.PYTHON_LIST_ARG_PREFIX) or \
      input_wstl_arg.startswith(_constants.FILE_ARG_PREFIX):
    if not load_contents:
      return [
          wstlservice_pb2.Location(local_path=inline_json)
          for inline_json in inputs
      ]
    else:
      return [
          wstlservice_pb2.Location(inline_json=inline_json)
          for inline_json in inputs
      ]
  elif input_wstl_arg.startswith(_constants.GS_ARG_PREFIX):
    return [
        wstlservice_pb2.Location(gcs_location=gcs_path)
        for gcs_path in gcs_paths
    ]
  else:
    raise ValueError("Missing {} supported prefix".format(",".join([
        _constants.JSON_ARG_PREFIX, _constants.GS_ARG_PREFIX,
        _constants.PYTHON_ARG_PREFIX, _constants.PYTHON_LIST_ARG_PREFIX
    ])))