Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/c7n_gcp/resources/dataflow.py: 62%

1# Copyright The Cloud Custodian Authors. 

2# SPDX-License-Identifier: Apache-2.0 

3from googleapiclient.errors import HttpError 

4 

5from c7n_gcp.provider import resources 

6from c7n_gcp.query import QueryResourceManager, TypeInfo 

7from c7n.utils import jmespath_search 

8 

9 

10@resources.register('dataflow-job') 

11class DataflowJob(QueryResourceManager): 

12 """GCP resource: https://cloud.google.com/dataflow/docs/reference/rest/v1b3/projects.jobs 

13 """ 

14 

15 class resource_type(TypeInfo): 

16 service = 'dataflow' 

17 version = 'v1b3' 

18 component = 'projects.jobs' 

19 enum_spec = ('aggregated', 'jobs[]', None) 

20 scope_key = 'projectId' 

21 name = id = 'name' 

22 get_requires_event = True 

23 default_report_fields = [ 

24 'name', 'currentState', 'createTime', 'location'] 

25 permissions = ('dataflow.jobs.list',) 

26 urn_component = "job" 

27 urn_region_key = 'location' 

28 asset_type = "dataflow.googleapis.com/Job" 

29 

30 @staticmethod 

31 def get(client, event): 

32 return client.execute_command( 

33 'get', { 

34 'projectId': jmespath_search('resource.labels.project_id', event), 

35 'jobId': jmespath_search('protoPayload.request.job_id', event) 

36 } 

37 ) 

38 

39 def resources(self, query=None): 

40 query_filter = 'ACTIVE' 

41 if self.data.get('query'): 

42 query_filter = self.data['query'][0].get('filter', 'ACTIVE') 

43 

44 return super(DataflowJob, self).resources(query={'filter': query_filter}) 

45 

46 def augment(self, resources): 

47 client = self.get_client() 

48 results = [] 

49 for r in resources: 

50 ref = { 

51 'jobId': r['id'], 

52 'projectId': r['projectId'], 

53 'view': 'JOB_VIEW_ALL' 

54 } 

55 try: 

56 results.append( 

57 client.execute_query( 

58 'get', verb_arguments=ref)) 

59 except HttpError: 

60 results.append(r) 

61 return results