def get_api_stub()

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

• 89 lines of code
• 21 McCabe index (conditional complexity)

def get_api_stub(service_name: str,
                 version: str,
                 project_id: Optional[str] = None,
                 regional_service_endpoint: Optional[str] = None):

  # Avoid circular import dependencies by importing the required modules here.
  # pylint: disable=import-outside-toplevel

  if service_name == 'cloudresourcemanager':
    from gcpdiag.queries import crm_stub
    return crm_stub.CrmApiStub()
  elif service_name == 'container':
    from gcpdiag.queries import gke_stub
    return gke_stub.ContainerApiStub()
  elif service_name == 'cloudkms':
    from gcpdiag.queries import kms_stub
    return kms_stub.KmsApiStub()
  elif service_name == 'compute':
    from gcpdiag.queries import gce_stub
    return gce_stub.ComputeEngineApiStub()
  elif service_name == 'iam':
    from gcpdiag.queries import iam_stub

    # project_id isn't required in serviceaccounts.get requests
    # and it cannot be easily extracted from a service account name
    # thus passing api_project_id to determine the correct test-data
    return iam_stub.IamApiStub(project_id)
  elif service_name == 'logging':
    from gcpdiag.queries import logs_stub
    return logs_stub.LoggingApiStub()
  elif service_name == 'monitoring':
    from gcpdiag.queries import monitoring_stub
    return monitoring_stub.MonitoringApiStub()
  elif service_name == 'serviceusage':
    return ServiceUsageApiStub()
  elif service_name == 'cloudfunctions':
    from gcpdiag.queries import gcf_stub
    return gcf_stub.CloudFunctionsApiStub()
  elif service_name == 'datafusion':
    from gcpdiag.queries import datafusion_stub
    return datafusion_stub.DataFusionApiStub()
  elif service_name == 'dataproc':
    from gcpdiag.queries import dataproc_stub
    return dataproc_stub.DataprocApiStub()
  elif service_name == 'apigee':
    from gcpdiag.queries import apigee_stub
    return apigee_stub.ApigeeApiStub()
  elif service_name == 'composer':
    from gcpdiag.queries import composer_stub
    return composer_stub.ComposerApiStub()
  elif service_name == 'sqladmin':
    from gcpdiag.queries import cloudsql_stub
    return cloudsql_stub.CloudSQLApiStub()
  elif service_name == 'storage':
    from gcpdiag.queries import gcs_stub

    # project_id isn't required in buckets.get and buckets.getIamPolicy requests
    # and it cannot be extracted from bucket name
    # thus passing api_project_id to determine the correct test-data
    return gcs_stub.BucketApiStub(project_id)
  elif service_name == 'cloudbuild':
    from gcpdiag.queries import gcb_stub
    return gcb_stub.CloudBuildApiStub()
  elif service_name == 'pubsub':
    from gcpdiag.queries import pubsub_stub
    return pubsub_stub.PubsubApiStub()
  elif service_name == 'appengine':
    from gcpdiag.queries import gae_stub
    return gae_stub.AppEngineApiStub()
  elif service_name == 'artifactregistry':
    from gcpdiag.queries import artifact_registry_stub
    return artifact_registry_stub.ArtifactRegistryApiStub()
  elif service_name == 'run':
    from gcpdiag.queries import cloudrun_stub
    return cloudrun_stub.CloudRunApiStub()
  elif service_name == 'notebooks':
    from gcpdiag.queries import notebooks_stub
    return notebooks_stub.NotebooksApiStub()
  elif service_name == 'dataflow':
    from gcpdiag.queries import dataflow_stub
    return dataflow_stub.DataflowApiStub()
  elif 'aiplatform' in service_name:
    from gcpdiag.queries import vertex_stub
    return vertex_stub.VertexApiStub()
  elif service_name == 'recommender':
    from gcpdiag.queries import recommender_stub
    return recommender_stub.RecommenderApiStub()
  elif service_name == 'cloudbilling':
    from gcpdiag.queries import billing_stub
    return billing_stub.BillingApiStub()
  elif service_name == 'osconfig':
    from gcpdiag.queries import osconfig_stub
    return osconfig_stub.OSConfigStub()
  elif service_name == 'networkmanagement':
    from gcpdiag.queries import networkmanagement_stub
    return networkmanagement_stub.NetworkManagementApiStub()
  elif service_name == 'cloudasset':
    from gcpdiag.queries import cloudasset_stub
    return cloudasset_stub.CloudAssetApiStub()
  else:
    raise ValueError('unsupported service: %s' % service_name)