HEX
Server: Apache/2.4.65 (Ubuntu)
System: Linux ielts-store-v2 6.8.0-1036-gcp #38~22.04.1-Ubuntu SMP Thu Aug 14 01:19:18 UTC 2025 x86_64
User: root (0)
PHP: 7.2.34-54+ubuntu20.04.1+deb.sury.org+1
Disabled: pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,
Upload Files
File: //snap/google-cloud-cli/current/lib/googlecloudsdk/command_lib/storage/symlink_util.py
# -*- coding: utf-8 -*- #
# Copyright 2023 Google LLC. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Symlink utilities for storage commands."""

from __future__ import absolute_import
from __future__ import division
from __future__ import unicode_literals

import os

from googlecloudsdk.command_lib.storage import tracker_file_util
from googlecloudsdk.command_lib.storage.resources import resource_util
from googlecloudsdk.core import properties
from googlecloudsdk.core.util import files


def update_custom_metadata_dict_with_symlink_attributes(
    metadata_dict, is_symlink
):
  """Updates custom metadata_dict with symlink data."""
  if is_symlink:
    metadata_dict[resource_util.SYMLINK_METADATA_KEY] = 'true'
  elif resource_util.SYMLINK_METADATA_KEY in metadata_dict:
    del metadata_dict[resource_util.SYMLINK_METADATA_KEY]


def _create_symlink_directory_if_needed():
  """Looks up or creates the gcloud storage symlink file directory.

  Symlink placeholder files will be kept here.

  Returns:
    The path string to the symlink directory.
  """
  symlink_directory = (
      properties.VALUES.storage.symlink_placeholder_directory.Get()
  )
  # Thread-safe method to prevent parallel processing errors.
  files.MakeDir(symlink_directory)
  return symlink_directory


def get_symlink_placeholder_path(source_path):
  """Returns a path suitable for storing a placeholder file for a symlink."""
  symlink_directory = _create_symlink_directory_if_needed()
  symlink_filename = tracker_file_util.get_hashed_file_name(
      tracker_file_util.get_delimiterless_file_path(source_path)
  )
  tracker_file_util.raise_exceeds_max_length_error(symlink_filename)
  return os.path.join(symlink_directory, symlink_filename)


def get_symlink_placeholder_file(source_path):
  """Creates a placholder file for the given symlink.

  The placeholder will be created in the directory specified by the
  symlink_placeholder_directory property, and its content will be the path
  targeted by the given symlink.

  Args:
    source_path: The path to an existing symlink for which a placeholder should
      be created.

  Returns:
    The path to the placeholder file that was created to represent the given
    symlink.
  """
  placeholder_path = get_symlink_placeholder_path(source_path)
  with files.BinaryFileWriter(placeholder_path) as placeholder_writer:
    placeholder_writer.write(os.readlink(source_path).encode('utf-8'))
  return placeholder_path


def create_symlink_from_temporary_placeholder(placeholder_path, symlink_path):
  symlink_target = files.ReadFileContents(placeholder_path)
  os.symlink(symlink_target, symlink_path)


def get_preserve_symlink_from_user_request(user_request_args):
  return user_request_args.preserve_symlinks if user_request_args else None