We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
GCP- dataflow deprecated view and now doesn't return complete information
Looks like gcp deprecated the dataflow list ability to get all dataflow job information. It only returns `JOB_VIEW_SUMMARY `

https://cloud.google.com/dataflow/docs/reference/rest/v1b3/projects.jobs/list

In order to get the required information a get call on every resource will be required. Should this be done on the resource level as an augmentation or in an additional filter since the jobs could be filtered on state in a value filter before the extra api calls. 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `tools/c7n_gcp/c7n_gcp/resources/dataflow.py`
Content:
```
1 # Copyright The Cloud Custodian Authors.
2 # SPDX-License-Identifier: Apache-2.0
3 import jmespath
4 
5 from c7n_gcp.provider import resources
6 from c7n_gcp.query import QueryResourceManager, TypeInfo
7 
8 
9 @resources.register('dataflow-job')
10 class DataflowJob(QueryResourceManager):
11     """GCP resource: https://cloud.google.com/dataflow/docs/reference/rest/v1b3/projects.jobs
12     """
13 
14     class resource_type(TypeInfo):
15         service = 'dataflow'
16         version = 'v1b3'
17         component = 'projects.jobs'
18         enum_spec = ('aggregated', 'jobs[]', None)
19         scope_key = 'projectId'
20         name = id = 'name'
21         get_requires_event = True
22         default_report_fields = [
23             'name', 'currentState', 'createTime', 'location']
24         permissions = ('dataflow.jobs.list',)
25 
26         @staticmethod
27         def get(client, event):
28             return client.execute_command(
29                 'get', {
30                     'projectId': jmespath.search('resource.labels.project_id', event),
31                     'jobId': jmespath.search('protoPayload.request.job_id', event)
32                 }
33             )
34
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/tools/c7n_gcp/c7n_gcp/resources/dataflow.py b/tools/c7n_gcp/c7n_gcp/resources/dataflow.py
--- a/tools/c7n_gcp/c7n_gcp/resources/dataflow.py
+++ b/tools/c7n_gcp/c7n_gcp/resources/dataflow.py
@@ -1,6 +1,7 @@
 # Copyright The Cloud Custodian Authors.
 # SPDX-License-Identifier: Apache-2.0
 import jmespath
+from googleapiclient.errors import HttpError
 
 from c7n_gcp.provider import resources
 from c7n_gcp.query import QueryResourceManager, TypeInfo
@@ -31,3 +32,27 @@
                     'jobId': jmespath.search('protoPayload.request.job_id', event)
                 }
             )
+
+    def resources(self, query=None):
+        query_filter = 'ACTIVE'
+        if self.data.get('query'):
+            query_filter = self.data['query'][0].get('filter', 'ACTIVE')
+
+        return super(DataflowJob, self).resources(query={'filter': query_filter})
+
+    def augment(self, resources):
+        client = self.get_client()
+        results = []
+        for r in resources:
+            ref = {
+                'jobId': r['id'],
+                'projectId': r['projectId'],
+                'view': 'JOB_VIEW_ALL'
+            }
+            try:
+                results.append(
+                    client.execute_query(
+                        'get', verb_arguments=ref))
+            except HttpError:
+                results.append(r)
+        return results

{"golden_diff": "diff --git a/tools/c7n_gcp/c7n_gcp/resources/dataflow.py b/tools/c7n_gcp/c7n_gcp/resources/dataflow.py\n--- a/tools/c7n_gcp/c7n_gcp/resources/dataflow.py\n+++ b/tools/c7n_gcp/c7n_gcp/resources/dataflow.py\n@@ -1,6 +1,7 @@\n # Copyright The Cloud Custodian Authors.\n # SPDX-License-Identifier: Apache-2.0\n import jmespath\n+from googleapiclient.errors import HttpError\n \n from c7n_gcp.provider import resources\n from c7n_gcp.query import QueryResourceManager, TypeInfo\n@@ -31,3 +32,27 @@\n                     'jobId': jmespath.search('protoPayload.request.job_id', event)\n                 }\n             )\n+\n+    def resources(self, query=None):\n+        query_filter = 'ACTIVE'\n+        if self.data.get('query'):\n+            query_filter = self.data['query'][0].get('filter', 'ACTIVE')\n+\n+        return super(DataflowJob, self).resources(query={'filter': query_filter})\n+\n+    def augment(self, resources):\n+        client = self.get_client()\n+        results = []\n+        for r in resources:\n+            ref = {\n+                'jobId': r['id'],\n+                'projectId': r['projectId'],\n+                'view': 'JOB_VIEW_ALL'\n+            }\n+            try:\n+                results.append(\n+                    client.execute_query(\n+                        'get', verb_arguments=ref))\n+            except HttpError:\n+                results.append(r)\n+        return results\n", "issue": "GCP- dataflow deprecated view and now doesn't return complete information\nLooks like gcp deprecated the dataflow list ability to get all dataflow job information. It only returns `JOB_VIEW_SUMMARY `\r\n\r\nhttps://cloud.google.com/dataflow/docs/reference/rest/v1b3/projects.jobs/list\r\n\r\nIn order to get the required information a get call on every resource will be required. Should this be done on the resource level as an augmentation or in an additional filter since the jobs could be filtered on state in a value filter before the extra api calls. \n", "before_files": [{"content": "# Copyright The Cloud Custodian Authors.\n# SPDX-License-Identifier: Apache-2.0\nimport jmespath\n\nfrom c7n_gcp.provider import resources\nfrom c7n_gcp.query import QueryResourceManager, TypeInfo\n\n\n@resources.register('dataflow-job')\nclass DataflowJob(QueryResourceManager):\n    \"\"\"GCP resource: https://cloud.google.com/dataflow/docs/reference/rest/v1b3/projects.jobs\n    \"\"\"\n\n    class resource_type(TypeInfo):\n        service = 'dataflow'\n        version = 'v1b3'\n        component = 'projects.jobs'\n        enum_spec = ('aggregated', 'jobs[]', None)\n        scope_key = 'projectId'\n        name = id = 'name'\n        get_requires_event = True\n        default_report_fields = [\n            'name', 'currentState', 'createTime', 'location']\n        permissions = ('dataflow.jobs.list',)\n\n        @staticmethod\n        def get(client, event):\n            return client.execute_command(\n                'get', {\n                    'projectId': jmespath.search('resource.labels.project_id', event),\n                    'jobId': jmespath.search('protoPayload.request.job_id', event)\n                }\n            )\n", "path": "tools/c7n_gcp/c7n_gcp/resources/dataflow.py"}], "after_files": [{"content": "# Copyright The Cloud Custodian Authors.\n# SPDX-License-Identifier: Apache-2.0\nimport jmespath\nfrom googleapiclient.errors import HttpError\n\nfrom c7n_gcp.provider import resources\nfrom c7n_gcp.query import QueryResourceManager, TypeInfo\n\n\n@resources.register('dataflow-job')\nclass DataflowJob(QueryResourceManager):\n    \"\"\"GCP resource: https://cloud.google.com/dataflow/docs/reference/rest/v1b3/projects.jobs\n    \"\"\"\n\n    class resource_type(TypeInfo):\n        service = 'dataflow'\n        version = 'v1b3'\n        component = 'projects.jobs'\n        enum_spec = ('aggregated', 'jobs[]', None)\n        scope_key = 'projectId'\n        name = id = 'name'\n        get_requires_event = True\n        default_report_fields = [\n            'name', 'currentState', 'createTime', 'location']\n        permissions = ('dataflow.jobs.list',)\n\n        @staticmethod\n        def get(client, event):\n            return client.execute_command(\n                'get', {\n                    'projectId': jmespath.search('resource.labels.project_id', event),\n                    'jobId': jmespath.search('protoPayload.request.job_id', event)\n                }\n            )\n\n    def resources(self, query=None):\n        query_filter = 'ACTIVE'\n        if self.data.get('query'):\n            query_filter = self.data['query'][0].get('filter', 'ACTIVE')\n\n        return super(DataflowJob, self).resources(query={'filter': query_filter})\n\n    def augment(self, resources):\n        client = self.get_client()\n        results = []\n        for r in resources:\n            ref = {\n                'jobId': r['id'],\n                'projectId': r['projectId'],\n                'view': 'JOB_VIEW_ALL'\n            }\n            try:\n                results.append(\n                    client.execute_query(\n                        'get', verb_arguments=ref))\n            except HttpError:\n                results.append(r)\n        return results\n", "path": "tools/c7n_gcp/c7n_gcp/resources/dataflow.py"}]}