def get_target_proxy_reference()

in gcpdiag/queries/lb.py [0:0]

• 54 lines of code
• 13 McCabe index (conditional complexity)

def get_target_proxy_reference(target_proxy_self_link: str) -> str:
  """Retrieves the URL map or backend service associated with a given target proxy.

  Args:
    target_proxy_self_link: self link of the target proxy

  Returns:
    The url map or the backend service self link
  """
  target_proxy_type = target_proxy_self_link.split('/')[-2]
  target_proxy_name = target_proxy_self_link.split('/')[-1]
  target_proxy_scope = target_proxy_self_link.split('/')[-3]
  match_result = re.match(r'projects/([^/]+)/', target_proxy_self_link)
  if not match_result:
    return ''
  project_id = match_result.group(1)
  compute = apis.get_api('compute', 'v1', project_id)

  request = None
  if target_proxy_type == 'targetHttpsProxies':
    if target_proxy_scope == 'global':
      request = compute.targetHttpsProxies().get(
          project=project_id, targetHttpsProxy=target_proxy_name)
    else:
      request = compute.regionTargetHttpsProxies().get(
          project=project_id,
          region=target_proxy_scope,
          targetHttpsProxy=target_proxy_name,
      )
  elif target_proxy_type == 'targetHttpProxies':
    if target_proxy_scope == 'global':
      request = compute.targetHttpProxies().get(
          project=project_id, targetHttpProxy=target_proxy_name)
    else:
      request = compute.regionTargetHttpProxies().get(
          project=project_id,
          region=target_proxy_scope,
          targetHttpProxy=target_proxy_name,
      )
  elif target_proxy_type == 'targetTcpProxies':
    if target_proxy_scope == 'global':
      request = compute.targetTcpProxies().get(project=project_id,
                                               targetTcpProxy=target_proxy_name)
    else:
      request = compute.regionTargetTcpProxies().get(
          project=project_id,
          region=target_proxy_scope,
          targetTcpProxy=target_proxy_name,
      )
  elif target_proxy_type == 'targetSslProxies':
    request = compute.targetSslProxies().get(project=project_id,
                                             targetSslProxy=target_proxy_name)
  elif target_proxy_type == 'targetGrcpProxies':
    request = compute.targetGrpcProxies().get(project=project_id,
                                              targetGrpcProxy=target_proxy_name)
  if not request:
    # target is not target proxy
    return ''
  response = request.execute(num_retries=config.API_RETRIES)
  if 'urlMap' in response:
    return normalize_url(response['urlMap'])
  if 'service' in response:
    return normalize_url(response['service'])
  return ''