We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Asset exporter tool - getting ImportError in GAE 
Just tried to set up from scratch in new project. Followed steps from readme
When running the cron job I get this

ImportError: cannot import name 'expr_pb2' from 'google.type' (/env/lib/python3.7/site-packages/google/type/__init__.py)
at <module> (/env/lib/python3.7/site-packages/google/iam/v1/policy_pb2.py:16)
at <module> (/env/lib/python3.7/site-packages/google/iam/v1/iam_policy_pb2.py:17)
at <module> (/env/lib/python3.7/site-packages/google/cloud/asset_v1/proto/assets_pb2.py:19)
at <module> (/env/lib/python3.7/site-packages/google/cloud/asset_v1/proto/asset_service_pb2.py:20)
at <module> (/env/lib/python3.7/site-packages/google/cloud/asset_v1/types.py:23)
at <module> (/env/lib/python3.7/site-packages/google/cloud/asset_v1/__init__.py:20)
at <module> (/srv/lib/asset_inventory/export.py:33)
at <module> (/srv/main.py:45)
at import_app (/env/lib/python3.7/site-packages/gunicorn/util.py:350)
at load_wsgiapp (/env/lib/python3.7/site-packages/gunicorn/app/wsgiapp.py:41)
at load (/env/lib/python3.7/site-packages/gunicorn/app/wsgiapp.py:52)
at wsgi (/env/lib/python3.7/site-packages/gunicorn/app/base.py:67)
at load_wsgi (/env/lib/python3.7/site-packages/gunicorn/workers/base.py:138)
at init_process (/env/lib/python3.7/site-packages/gunicorn/workers/base.py:129)
at init_process (/env/lib/python3.7/site-packages/gunicorn/workers/gthread.py:104)
at spawn_worker (/env/lib/python3.7/site-packages/gunicorn/arbiter.py:583)

--- 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/asset-inventory/asset_inventory/pipeline_runner.py`
Content:
```
1 # Copyright 2019 Google Inc.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 """Code to invoke the pipeline."""
15 
16 import logging
17 import pprint
18 import time
19 
20 from googleapiclient.discovery import build
21 from oauth2client.client import GoogleCredentials
22 
23 
24 def get_job_name(load_time):
25     """User friendly job name from load_time."""
26     return ('cloud-asset-import-' + load_time.lower().replace(
27         ':', '-').replace(' ', '').replace('.', '-'))
28 
29 
30 def is_successful_state(final_state):
31     """True if the status is successful.
32 
33     Checks both for beam and template runner success codes.
34 
35     Args:
36         final_state: Final state the pipeline is in.
37 
38     Returns:
39         True if the job was successful.
40     """
41     if final_state not in ['JOB_STATE_DONE', 'DONE']:
42         return False
43     return True
44 
45 
46 def wait_on_pipeline_job(df_service, pipeline_job):
47     """Poll the job status every 60 seconds until done."""
48     dataflow_project = pipeline_job['projectId']
49     template_region = pipeline_job['location']
50     job_id = pipeline_job['id']
51     pipeline_job = df_service.projects().locations().jobs().get(
52         location=template_region, projectId=dataflow_project,
53         jobId=job_id).execute(num_retries=5)
54     logging.info('job status %s', pprint.pformat(pipeline_job))
55     current_state = pipeline_job['currentState']
56     # We have reached a terminal state.
57     if current_state in [
58         'JOB_STATE_DONE', 'JOB_STATE_FAILED', 'JOB_STATE_CANCELLED',
59         'JOB_STATE_UPDATED', 'JOB_STATE_DRAINED'
60     ]:
61         logging.info('final pipeline state : %s', current_state)
62         return current_state, pipeline_job
63     logging.info('sleeping 60 seconds before repolling.')
64     time.sleep(60)
65     return wait_on_pipeline_job(df_service, pipeline_job)
66 
67 
68 def run_pipeline_template(dataflow_project, template_region, template_location,
69                           input_location, group_by, write_disposition, dataset,
70                           stage, load_time, num_shards, runtime_environment):
71     """Invoke the suplied pipeline template.
72 
73     Args:
74         dataflow_project: Project to run the dataflow job in.
75         template_region: Region to run the job in.
76         template_location: GCS path to the template file.
77         input_location: GCS path load json documents from,
78         group_by: How to split assets into tables.
79         write_disposition: To append to or ovewrite BigQuery tables.
80         dataset: BigQuery dataset to write to.
81         stage: GCS path to write BigQuery load files.
82         load_time: Timestamp or date to load data with.
83         num_shards: Shards for for each asset type.
84         runtime_environment: Dict  suppling other runtime overrides.
85     Returns:
86         End state of the pipline and job object.
87     """
88     credentials = GoogleCredentials.get_application_default()
89     df_service = build('dataflow', 'v1b3', credentials=credentials)
90 
91     # Set the following variables to your values.
92     job_name = get_job_name(load_time)
93     body = {
94         'jobName': job_name,
95         'parameters': {
96             'input': input_location,
97             'load_time': load_time,
98             'stage': stage,
99             'group_by': group_by,
100             'write_disposition': write_disposition,
101             'num_shards': num_shards,
102             'dataset': dataset,
103         },
104         'environment': runtime_environment
105     }
106     logging.info('launching template %s in %s:%s with %s', template_location,
107                  dataflow_project, template_region, pprint.pformat(body))
108     launch_result = df_service.projects().locations().templates().launch(
109         location=template_region,
110         projectId=dataflow_project,
111         gcsPath=template_location,
112         body=body).execute(num_retries=5)
113 
114     logging.info('waiting on pipeline : %s', pprint.pformat(launch_result))
115     return wait_on_pipeline_job(df_service, launch_result['job'])
116 
117 
118 def run_pipeline_beam_runner(pipeline_runner, dataflow_project, input_location,
119                              group_by, write_disposition, dataset, stage,
120                              load_time, num_shards, pipeline_arguments):
121     """Invokes the pipeline with a beam runner.
122 
123     Only tested with the dataflow and direct runners.
124 
125     Args:
126         pipeline_runner: The Beam runner to use.
127         dataflow_project: Project to run the dataflow job in.
128         input_location: GCS path load json documents from,
129         group_by: How to split assets into tables.
130         write_disposition: To append to or ovewrite BigQuery tables.
131         dataset: BigQuery dataset to write to.
132         stage: GCS path to write BigQuery load files.
133         load_time: Timestamp to add to data during during BigQuery load.
134         num_shards: Shards for for each asset type.
135         pipeline_arguments: List of additional runner arguments.
136     Returns:
137         The end state of the pipeline run (a string), and PipelineResult.
138     """
139 
140     # pylint: disable=import-error
141     # import on demand as we don't want to depend on pipeline code which imports
142     # apache beam code unless we are using a beam runner and not invoking a
143     # template.
144     from asset_inventory import import_pipeline
145     job_name = get_job_name(load_time)
146 
147     pipeline_parameters = pipeline_arguments
148 
149     parameters = {
150         '--load_time': load_time,
151         '--job_name': job_name,
152         '--project': dataflow_project,
153         '--input': input_location,
154         '--group_by': group_by,
155         '--write_disposition': write_disposition,
156         '--num_shards': num_shards,
157         '--dataset': dataset,
158         '--stage': stage,
159         '--runner': pipeline_runner
160     }
161     for arg_name, value in parameters.items():
162         if value and arg_name not in pipeline_parameters:
163             pipeline_parameters += [arg_name, value]
164     pipeline_result = import_pipeline.run(pipeline_parameters)
165     logging.info('waiting on pipeline : %s', pprint.pformat(pipeline_result))
166     state = pipeline_result.wait_until_finish()
167     logging.info('final pipeline state: %s', state)
168     return pipeline_result.state, pipeline_result
169
```
--- 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/asset-inventory/asset_inventory/pipeline_runner.py b/tools/asset-inventory/asset_inventory/pipeline_runner.py
--- a/tools/asset-inventory/asset_inventory/pipeline_runner.py
+++ b/tools/asset-inventory/asset_inventory/pipeline_runner.py
@@ -86,7 +86,8 @@
         End state of the pipline and job object.
     """
     credentials = GoogleCredentials.get_application_default()
-    df_service = build('dataflow', 'v1b3', credentials=credentials)
+    df_service = build('dataflow', 'v1b3', credentials=credentials,
+                       cache_discovery=False)
 
     # Set the following variables to your values.
     job_name = get_job_name(load_time)

{"golden_diff": "diff --git a/tools/asset-inventory/asset_inventory/pipeline_runner.py b/tools/asset-inventory/asset_inventory/pipeline_runner.py\n--- a/tools/asset-inventory/asset_inventory/pipeline_runner.py\n+++ b/tools/asset-inventory/asset_inventory/pipeline_runner.py\n@@ -86,7 +86,8 @@\n         End state of the pipline and job object.\n     \"\"\"\n     credentials = GoogleCredentials.get_application_default()\n-    df_service = build('dataflow', 'v1b3', credentials=credentials)\n+    df_service = build('dataflow', 'v1b3', credentials=credentials,\n+                       cache_discovery=False)\n \n     # Set the following variables to your values.\n     job_name = get_job_name(load_time)\n", "issue": "Asset exporter tool - getting ImportError in GAE \nJust tried to set up from scratch in new project. Followed steps from readme\r\nWhen running the cron job I get this\r\n\r\nImportError: cannot import name 'expr_pb2' from 'google.type' (/env/lib/python3.7/site-packages/google/type/__init__.py)\r\nat <module> (/env/lib/python3.7/site-packages/google/iam/v1/policy_pb2.py:16)\r\nat <module> (/env/lib/python3.7/site-packages/google/iam/v1/iam_policy_pb2.py:17)\r\nat <module> (/env/lib/python3.7/site-packages/google/cloud/asset_v1/proto/assets_pb2.py:19)\r\nat <module> (/env/lib/python3.7/site-packages/google/cloud/asset_v1/proto/asset_service_pb2.py:20)\r\nat <module> (/env/lib/python3.7/site-packages/google/cloud/asset_v1/types.py:23)\r\nat <module> (/env/lib/python3.7/site-packages/google/cloud/asset_v1/__init__.py:20)\r\nat <module> (/srv/lib/asset_inventory/export.py:33)\r\nat <module> (/srv/main.py:45)\r\nat import_app (/env/lib/python3.7/site-packages/gunicorn/util.py:350)\r\nat load_wsgiapp (/env/lib/python3.7/site-packages/gunicorn/app/wsgiapp.py:41)\r\nat load (/env/lib/python3.7/site-packages/gunicorn/app/wsgiapp.py:52)\r\nat wsgi (/env/lib/python3.7/site-packages/gunicorn/app/base.py:67)\r\nat load_wsgi (/env/lib/python3.7/site-packages/gunicorn/workers/base.py:138)\r\nat init_process (/env/lib/python3.7/site-packages/gunicorn/workers/base.py:129)\r\nat init_process (/env/lib/python3.7/site-packages/gunicorn/workers/gthread.py:104)\r\nat spawn_worker (/env/lib/python3.7/site-packages/gunicorn/arbiter.py:583)\n", "before_files": [{"content": "# Copyright 2019 Google Inc.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Code to invoke the pipeline.\"\"\"\n\nimport logging\nimport pprint\nimport time\n\nfrom googleapiclient.discovery import build\nfrom oauth2client.client import GoogleCredentials\n\n\ndef get_job_name(load_time):\n    \"\"\"User friendly job name from load_time.\"\"\"\n    return ('cloud-asset-import-' + load_time.lower().replace(\n        ':', '-').replace(' ', '').replace('.', '-'))\n\n\ndef is_successful_state(final_state):\n    \"\"\"True if the status is successful.\n\n    Checks both for beam and template runner success codes.\n\n    Args:\n        final_state: Final state the pipeline is in.\n\n    Returns:\n        True if the job was successful.\n    \"\"\"\n    if final_state not in ['JOB_STATE_DONE', 'DONE']:\n        return False\n    return True\n\n\ndef wait_on_pipeline_job(df_service, pipeline_job):\n    \"\"\"Poll the job status every 60 seconds until done.\"\"\"\n    dataflow_project = pipeline_job['projectId']\n    template_region = pipeline_job['location']\n    job_id = pipeline_job['id']\n    pipeline_job = df_service.projects().locations().jobs().get(\n        location=template_region, projectId=dataflow_project,\n        jobId=job_id).execute(num_retries=5)\n    logging.info('job status %s', pprint.pformat(pipeline_job))\n    current_state = pipeline_job['currentState']\n    # We have reached a terminal state.\n    if current_state in [\n        'JOB_STATE_DONE', 'JOB_STATE_FAILED', 'JOB_STATE_CANCELLED',\n        'JOB_STATE_UPDATED', 'JOB_STATE_DRAINED'\n    ]:\n        logging.info('final pipeline state : %s', current_state)\n        return current_state, pipeline_job\n    logging.info('sleeping 60 seconds before repolling.')\n    time.sleep(60)\n    return wait_on_pipeline_job(df_service, pipeline_job)\n\n\ndef run_pipeline_template(dataflow_project, template_region, template_location,\n                          input_location, group_by, write_disposition, dataset,\n                          stage, load_time, num_shards, runtime_environment):\n    \"\"\"Invoke the suplied pipeline template.\n\n    Args:\n        dataflow_project: Project to run the dataflow job in.\n        template_region: Region to run the job in.\n        template_location: GCS path to the template file.\n        input_location: GCS path load json documents from,\n        group_by: How to split assets into tables.\n        write_disposition: To append to or ovewrite BigQuery tables.\n        dataset: BigQuery dataset to write to.\n        stage: GCS path to write BigQuery load files.\n        load_time: Timestamp or date to load data with.\n        num_shards: Shards for for each asset type.\n        runtime_environment: Dict  suppling other runtime overrides.\n    Returns:\n        End state of the pipline and job object.\n    \"\"\"\n    credentials = GoogleCredentials.get_application_default()\n    df_service = build('dataflow', 'v1b3', credentials=credentials)\n\n    # Set the following variables to your values.\n    job_name = get_job_name(load_time)\n    body = {\n        'jobName': job_name,\n        'parameters': {\n            'input': input_location,\n            'load_time': load_time,\n            'stage': stage,\n            'group_by': group_by,\n            'write_disposition': write_disposition,\n            'num_shards': num_shards,\n            'dataset': dataset,\n        },\n        'environment': runtime_environment\n    }\n    logging.info('launching template %s in %s:%s with %s', template_location,\n                 dataflow_project, template_region, pprint.pformat(body))\n    launch_result = df_service.projects().locations().templates().launch(\n        location=template_region,\n        projectId=dataflow_project,\n        gcsPath=template_location,\n        body=body).execute(num_retries=5)\n\n    logging.info('waiting on pipeline : %s', pprint.pformat(launch_result))\n    return wait_on_pipeline_job(df_service, launch_result['job'])\n\n\ndef run_pipeline_beam_runner(pipeline_runner, dataflow_project, input_location,\n                             group_by, write_disposition, dataset, stage,\n                             load_time, num_shards, pipeline_arguments):\n    \"\"\"Invokes the pipeline with a beam runner.\n\n    Only tested with the dataflow and direct runners.\n\n    Args:\n        pipeline_runner: The Beam runner to use.\n        dataflow_project: Project to run the dataflow job in.\n        input_location: GCS path load json documents from,\n        group_by: How to split assets into tables.\n        write_disposition: To append to or ovewrite BigQuery tables.\n        dataset: BigQuery dataset to write to.\n        stage: GCS path to write BigQuery load files.\n        load_time: Timestamp to add to data during during BigQuery load.\n        num_shards: Shards for for each asset type.\n        pipeline_arguments: List of additional runner arguments.\n    Returns:\n        The end state of the pipeline run (a string), and PipelineResult.\n    \"\"\"\n\n    # pylint: disable=import-error\n    # import on demand as we don't want to depend on pipeline code which imports\n    # apache beam code unless we are using a beam runner and not invoking a\n    # template.\n    from asset_inventory import import_pipeline\n    job_name = get_job_name(load_time)\n\n    pipeline_parameters = pipeline_arguments\n\n    parameters = {\n        '--load_time': load_time,\n        '--job_name': job_name,\n        '--project': dataflow_project,\n        '--input': input_location,\n        '--group_by': group_by,\n        '--write_disposition': write_disposition,\n        '--num_shards': num_shards,\n        '--dataset': dataset,\n        '--stage': stage,\n        '--runner': pipeline_runner\n    }\n    for arg_name, value in parameters.items():\n        if value and arg_name not in pipeline_parameters:\n            pipeline_parameters += [arg_name, value]\n    pipeline_result = import_pipeline.run(pipeline_parameters)\n    logging.info('waiting on pipeline : %s', pprint.pformat(pipeline_result))\n    state = pipeline_result.wait_until_finish()\n    logging.info('final pipeline state: %s', state)\n    return pipeline_result.state, pipeline_result\n", "path": "tools/asset-inventory/asset_inventory/pipeline_runner.py"}], "after_files": [{"content": "# Copyright 2019 Google Inc.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Code to invoke the pipeline.\"\"\"\n\nimport logging\nimport pprint\nimport time\n\nfrom googleapiclient.discovery import build\nfrom oauth2client.client import GoogleCredentials\n\n\ndef get_job_name(load_time):\n    \"\"\"User friendly job name from load_time.\"\"\"\n    return ('cloud-asset-import-' + load_time.lower().replace(\n        ':', '-').replace(' ', '').replace('.', '-'))\n\n\ndef is_successful_state(final_state):\n    \"\"\"True if the status is successful.\n\n    Checks both for beam and template runner success codes.\n\n    Args:\n        final_state: Final state the pipeline is in.\n\n    Returns:\n        True if the job was successful.\n    \"\"\"\n    if final_state not in ['JOB_STATE_DONE', 'DONE']:\n        return False\n    return True\n\n\ndef wait_on_pipeline_job(df_service, pipeline_job):\n    \"\"\"Poll the job status every 60 seconds until done.\"\"\"\n    dataflow_project = pipeline_job['projectId']\n    template_region = pipeline_job['location']\n    job_id = pipeline_job['id']\n    pipeline_job = df_service.projects().locations().jobs().get(\n        location=template_region, projectId=dataflow_project,\n        jobId=job_id).execute(num_retries=5)\n    logging.info('job status %s', pprint.pformat(pipeline_job))\n    current_state = pipeline_job['currentState']\n    # We have reached a terminal state.\n    if current_state in [\n        'JOB_STATE_DONE', 'JOB_STATE_FAILED', 'JOB_STATE_CANCELLED',\n        'JOB_STATE_UPDATED', 'JOB_STATE_DRAINED'\n    ]:\n        logging.info('final pipeline state : %s', current_state)\n        return current_state, pipeline_job\n    logging.info('sleeping 60 seconds before repolling.')\n    time.sleep(60)\n    return wait_on_pipeline_job(df_service, pipeline_job)\n\n\ndef run_pipeline_template(dataflow_project, template_region, template_location,\n                          input_location, group_by, write_disposition, dataset,\n                          stage, load_time, num_shards, runtime_environment):\n    \"\"\"Invoke the suplied pipeline template.\n\n    Args:\n        dataflow_project: Project to run the dataflow job in.\n        template_region: Region to run the job in.\n        template_location: GCS path to the template file.\n        input_location: GCS path load json documents from,\n        group_by: How to split assets into tables.\n        write_disposition: To append to or ovewrite BigQuery tables.\n        dataset: BigQuery dataset to write to.\n        stage: GCS path to write BigQuery load files.\n        load_time: Timestamp or date to load data with.\n        num_shards: Shards for for each asset type.\n        runtime_environment: Dict  suppling other runtime overrides.\n    Returns:\n        End state of the pipline and job object.\n    \"\"\"\n    credentials = GoogleCredentials.get_application_default()\n    df_service = build('dataflow', 'v1b3', credentials=credentials,\n                       cache_discovery=False)\n\n    # Set the following variables to your values.\n    job_name = get_job_name(load_time)\n    body = {\n        'jobName': job_name,\n        'parameters': {\n            'input': input_location,\n            'load_time': load_time,\n            'stage': stage,\n            'group_by': group_by,\n            'write_disposition': write_disposition,\n            'num_shards': num_shards,\n            'dataset': dataset,\n        },\n        'environment': runtime_environment\n    }\n    logging.info('launching template %s in %s:%s with %s', template_location,\n                 dataflow_project, template_region, pprint.pformat(body))\n    launch_result = df_service.projects().locations().templates().launch(\n        location=template_region,\n        projectId=dataflow_project,\n        gcsPath=template_location,\n        body=body).execute(num_retries=5)\n\n    logging.info('waiting on pipeline : %s', pprint.pformat(launch_result))\n    return wait_on_pipeline_job(df_service, launch_result['job'])\n\n\ndef run_pipeline_beam_runner(pipeline_runner, dataflow_project, input_location,\n                             group_by, write_disposition, dataset, stage,\n                             load_time, num_shards, pipeline_arguments):\n    \"\"\"Invokes the pipeline with a beam runner.\n\n    Only tested with the dataflow and direct runners.\n\n    Args:\n        pipeline_runner: The Beam runner to use.\n        dataflow_project: Project to run the dataflow job in.\n        input_location: GCS path load json documents from,\n        group_by: How to split assets into tables.\n        write_disposition: To append to or ovewrite BigQuery tables.\n        dataset: BigQuery dataset to write to.\n        stage: GCS path to write BigQuery load files.\n        load_time: Timestamp to add to data during during BigQuery load.\n        num_shards: Shards for for each asset type.\n        pipeline_arguments: List of additional runner arguments.\n    Returns:\n        The end state of the pipeline run (a string), and PipelineResult.\n    \"\"\"\n\n    # pylint: disable=import-error\n    # import on demand as we don't want to depend on pipeline code which imports\n    # apache beam code unless we are using a beam runner and not invoking a\n    # template.\n    from asset_inventory import import_pipeline\n    job_name = get_job_name(load_time)\n\n    pipeline_parameters = pipeline_arguments\n\n    parameters = {\n        '--load_time': load_time,\n        '--job_name': job_name,\n        '--project': dataflow_project,\n        '--input': input_location,\n        '--group_by': group_by,\n        '--write_disposition': write_disposition,\n        '--num_shards': num_shards,\n        '--dataset': dataset,\n        '--stage': stage,\n        '--runner': pipeline_runner\n    }\n    for arg_name, value in parameters.items():\n        if value and arg_name not in pipeline_parameters:\n            pipeline_parameters += [arg_name, value]\n    pipeline_result = import_pipeline.run(pipeline_parameters)\n    logging.info('waiting on pipeline : %s', pprint.pformat(pipeline_result))\n    state = pipeline_result.wait_until_finish()\n    logging.info('final pipeline state: %s', state)\n    return pipeline_result.state, pipeline_result\n", "path": "tools/asset-inventory/asset_inventory/pipeline_runner.py"}]}