GnnForPrivacyScan/data/purposeSplit/Share/source_util_2.py

def SetFunctionSourceProps(function, function_ref, source_arg, stage_bucket):
  """Add sources to function.

  Args:
    function: The function to add a source to.
    function_ref: The reference to the function.
    source_arg: Location of source code to deploy.
    stage_bucket: The name of the Google Cloud Storage bucket where source code
        will be stored.
  Returns:
    A list of fields on the function that have been changed.
  """
  function.sourceArchiveUrl = None
  function.sourceRepository = None
  function.sourceUploadUrl = None

  messages = api_util.GetApiMessagesModule()

  if source_arg is None:
    source_arg = '.'
  source_arg = source_arg or '.'
  if source_arg.startswith('gs://'):
    if not source_arg.endswith('.zip'):
      # Users may have .zip archives with unusual names, and we don't want to
      # prevent those from being deployed; the deployment should go through so
      # just warn here.
      log.warning(
          '[{}] does not end with extension `.zip`. '
          'The `--source` argument must designate the zipped source archive '
          'when providing a Google Cloud Storage URI.'.format(source_arg))
    function.sourceArchiveUrl = source_arg
    return ['sourceArchiveUrl']
  elif source_arg.startswith('https://'):
    function.sourceRepository = messages.SourceRepository(
        url=_AddDefaultBranch(source_arg)
    )
    return ['sourceRepository']
  with file_utils.TemporaryDirectory() as tmp_dir:
    zip_file = _CreateSourcesZipFile(tmp_dir, source_arg)
    service = api_util.GetApiClientInstance().projects_locations_functions

    upload_url = UploadFile(
        zip_file, stage_bucket, messages, service, function_ref)
    if upload_url.startswith('gs://'):
      function.sourceArchiveUrl = upload_url
      return ['sourceArchiveUrl']
    else:
      function.sourceUploadUrl = upload_url
      return ['sourceUploadUrl']