gcpdiag.queries.recommender

26class Recommendation(models.Resource):
27  """Represents a Recommender Recommendation.
28
29  See also the API documentation:
30  https://cloud.google.com/recommender/docs/reference/rest/v1/projects.locations.recommenders.recommendations
31  """
32
33  def __init__(self, project_id, resource_data):
34    super().__init__(project_id=project_id)
35    self._resource_data = resource_data
36
37  @property
38  def name(self) -> str:
39    return self._resource_data['name']
40
41  @property
42  def full_path(self) -> str:
43    return self.name
44
45  @property
46  def description(self) -> str:
47    return self._resource_data.get('description', '')
48
49  @property
50  def recommender_subtype(self) -> str:
51    return self._resource_data.get('recommenderSubtype', '')
52
53  @property
54  def state(self) -> str:
55    return self._resource_data.get('stateInfo', {}).get('state', '')
56
57  @property
58  def target_resources(self) -> List[str]:
59    return self._resource_data.get('targetResources', [])
60
61  @property
62  def content(self) -> dict:
63    return self._resource_data.get('content', {})
64
65  @property
66  def primary_impact(self) -> dict:
67    return self._resource_data.get('primaryImpact', {})

70@caching.cached_api_call
71def get_recommendations(
72  context: models.Context,
73  recommender_id: str,
74  location: str = '-',
75) -> List[Recommendation]:
76  """Get a list of Recommendations for a given recommender in a project, caching the result."""
77  recommendations: List[Recommendation] = []
78  project_id = context.project_id
79  if not apis.is_enabled(project_id, 'recommender'):
80    return recommendations
81
82  api = apis.get_api('recommender', 'v1', project_id)
83  parent = f'projects/{project_id}/locations/{location}/recommenders/{recommender_id}'
84  logging.debug('fetching recommendations for %s in location %s', recommender_id, location)
85
86  try:
87    for rec in apis_utils.list_all(
88      request=api.projects().locations().recommenders().recommendations().list(parent=parent),
89      next_function=api.projects().locations().recommenders().recommendations().list_next,
90      response_keyword='recommendations',
91    ):
92      recommendations.append(Recommendation(project_id, rec))
93  except googleapiclient.errors.HttpError as err:
94    raise utils.GcpApiError(err) from err
95  return recommendations