gcpdiag.queries.cloudasset

29class AssetResource(models.Resource):
30  """Represents Resource Retrieved from the Cloud Asset Inventory."""
31
32  _resource_data: dict
33
34  def __init__(self, project_id, resource_data):
35    super().__init__(project_id=project_id)
36    self._resource_data = resource_data
37
38  @property
39  def name(self) -> str:
40    m = re.search(r'//(.+)', self._resource_data['name'])
41    if not m:
42      raise RuntimeError("can't determine name of service %s" % (self._resource_data['name']))
43    return m.group(1)
44
45  @property
46  def full_path(self) -> str:
47    return self.name
48
49  @property
50  def asset_type(self) -> str:
51    return self._resource_data['assetType']

54@caching.cached_api_call
55def search_all_resources(
56  project_id: str,
57  asset_type: Optional[str] = None,
58  query: Optional[str] = None,
59) -> Mapping[str, AssetResource]:
60  """Searches all resources in the project."""
61  resources: Dict[str, AssetResource] = {}
62
63  if not apis.is_enabled(project_id, 'cloudasset'):
64    return resources
65  cloudasset_api = apis.get_api('cloudasset', 'v1', project_id)
66  logging.debug('fetching list of resources in the project %s', project_id)
67  request = cloudasset_api.v1().searchAllResources(
68    scope=f'projects/{project_id}', assetTypes=asset_type, query=query
69  )
70  response = request.execute(num_retries=config.API_RETRIES)
71  try:
72    if 'results' in response:
73      for resource in response['results']:
74        resources[resource['name']] = AssetResource(project_id, resource)
75  except googleapiclient.errors.HttpError as err:
76    raise utils.GcpApiError(err) from err
77  return resources