We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
just a few typos fixed
That Python version check can be removed in a few days when Colab drops Python 2 support.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `tfx/examples/custom_components/slack/example/taxi_pipeline_slack_kubeflow.py`
Content:
```
1 # Lint as: python2, python3
2 # Copyright 2019 Google LLC. All Rights Reserved.
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 #     http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 """Example pipeline to demonstrate custom TFX component.
16 
17 This example consists of standard TFX components as well as a custom TFX
18 component requesting for manual review through Slack.
19 
20 This example along with the custom `SlackComponent` will only serve as an
21 example and will not be supported by TFX team.
22 
23 This example runs in Kubeflow with Google Cloud services..
24 """
25 
26 from __future__ import absolute_import
27 from __future__ import division
28 from __future__ import print_function
29 
30 import datetime
31 import os
32 
33 from slack_component.component import SlackComponent
34 
35 from tfx.components import CsvExampleGen
36 from tfx.components import Evaluator
37 from tfx.components import ExampleValidator
38 from tfx.components import ModelValidator
39 from tfx.components import Pusher
40 from tfx.components import SchemaGen
41 from tfx.components import StatisticsGen
42 from tfx.components import Trainer
43 from tfx.components import Transform
44 from tfx.orchestration import pipeline
45 from tfx.orchestration.kubeflow import kubeflow_dag_runner
46 from tfx.proto import evaluator_pb2
47 from tfx.proto import pusher_pb2
48 from tfx.proto import trainer_pb2
49 from tfx.utils.dsl_utils import csv_input
50 
51 # This example assumes that the taxi data is stored in _input_bucket/data/simple
52 # and the taxi utility function is in example/taxi_utils_slack.py.
53 # Feel free to customize this as needed.
54 _input_bucket = 'gs://my-bucket'
55 _output_bucket = 'gs://my-bucket'
56 _taxi_root = __file__
57 _data_root = os.path.join(_input_bucket, 'data', 'simple')
58 
59 
60 # Python module file to inject customized logic into the TFX components. The
61 # Transform and Trainer both require user-defined functions to run successfully.
62 _taxi_trainer_func = 'example.taxi_utils_slack.trainer_fn'
63 _taxi_transformer_func = 'example.taxi_utils_slack.preprocessing_fn'
64 # Path which can be listened to by the model server.  Pusher will output the
65 # trained model here.
66 _serving_model_dir = os.path.join(_taxi_root, 'serving_model/taxi_slack')
67 # Slack channel to push the model notifications to.
68 _slack_channel_id = 'my-channel-id'
69 # Slack token to set up connection.
70 _slack_token = os.environ['SLACK_BOT_TOKEN']
71 
72 # Directory and data locations.  This example assumes all of the chicago taxi
73 # example code and metadata library is relative to $HOME, but you can store
74 # these files anywhere on your local filesystem.
75 _tfx_root = os.path.join(os.environ['HOME'], '/tfx')
76 _pipeline_name = 'chicago_taxi_slack_kubeflow'
77 _pipeline_root = os.path.join(_input_bucket, _pipeline_name)
78 _log_root = os.path.join(_tfx_root, 'logs')
79 
80 # Airflow-specific configs; these will be passed directly to airflow
81 _airflow_config = {
82     'schedule_interval': None,
83     'start_date': datetime.datetime(2019, 1, 1),
84 }
85 
86 
87 def _create_pipeline():
88   """Implements the chicago taxi pipeline with TFX."""
89   examples = csv_input(_data_root)
90 
91   # Brings data into the pipeline or otherwise joins/converts training data.
92   example_gen = CsvExampleGen(input=examples)
93 
94   # Computes statistics over data for visualization and example validation.
95   statistics_gen = StatisticsGen(examples=example_gen.outputs['examples'])
96 
97   # Generates schema based on statistics files.
98   infer_schema = SchemaGen(statistics=statistics_gen.outputs['statistics'])
99 
100   # Performs anomaly detection based on statistics and data schema.
101   validate_stats = ExampleValidator(
102       statistics=statistics_gen.outputs['statistics'],
103       schema=infer_schema.outputs['schema'])
104 
105   # Performs transformations and feature engineering in training and serving.
106   transform = Transform(
107       examples=example_gen.outputs['examples'],
108       schema=infer_schema.outputs['schema'],
109       preprocessing_fn=_taxi_transformer_func)
110 
111   # Uses user-provided Python function that implements a model using TF-Learn.
112   trainer = Trainer(
113       trainer_fn=_taxi_trainer_func,
114       examples=transform.outputs['transformed_examples'],
115       schema=infer_schema.outputs['schema'],
116       transform_graph=transform.outputs['transform_graph'],
117       train_args=trainer_pb2.TrainArgs(num_steps=10000),
118       eval_args=trainer_pb2.EvalArgs(num_steps=5000))
119 
120   # Uses TFMA to compute a evaluation statistics over features of a model.
121   model_analyzer = Evaluator(
122       examples=example_gen.outputs['examples'],
123       model_exports=trainer.outputs['model'],
124       feature_slicing_spec=evaluator_pb2.FeatureSlicingSpec(specs=[
125           evaluator_pb2.SingleSlicingSpec(
126               column_for_slicing=['trip_start_hour'])
127       ]))
128 
129   # Performs quality validation of a candidate model (compared to a baseline).
130   model_validator = ModelValidator(
131       examples=example_gen.outputs['examples'], model=trainer.outputs['model'])
132 
133   # This custom component serves as a bridge between pipeline and human model
134   # reviewers to enable review-and-push workflow in model development cycle. It
135   # utilizes Slack API to send message to user-defined Slack channel with model
136   # URI info and wait for go / no-go decision from the same Slack channel:
137   #   * To approve the model, users need to reply the thread sent out by the bot
138   #     started by SlackComponent with 'lgtm' or 'approve'.
139   #   * To reject the model, users need to reply the thread sent out by the bot
140   #     started by SlackComponent with 'decline' or 'reject'.
141   slack_validator = SlackComponent(
142       model=trainer.outputs['model'],
143       model_blessing=model_validator.outputs['blessing'],
144       slack_token=_slack_token,
145       slack_channel_id=_slack_channel_id,
146       timeout_sec=3600,
147   )
148  # Checks whether the model passed the validation steps and pushes the model
149   # to a file destination if check passed.
150   pusher = Pusher(
151       model=trainer.outputs['model'],
152       model_blessing=slack_validator.outputs['slack_blessing'],
153       push_destination=pusher_pb2.PushDestination(
154           filesystem=pusher_pb2.PushDestination.Filesystem(
155               base_directory=_serving_model_dir)))
156 
157   return pipeline.Pipeline(
158       pipeline_name=_pipeline_name,
159       pipeline_root=_pipeline_root,
160       components=[
161           example_gen, statistics_gen, infer_schema, validate_stats, transform,
162           trainer, model_analyzer, model_validator, slack_validator, pusher
163       ],
164       enable_cache=True,
165   )
166 
167 
168 if __name__ == '__main__':
169   # Metadata config. The defaults works work with the installation of
170   # KF Pipelines using Kubeflow. If installing KF Pipelines using the
171   # lightweight deployment option, you may need to override the defaults.
172   metadata_config = kubeflow_dag_runner.get_default_kubeflow_metadata_config()
173 
174   # This pipeline automatically injects the Kubeflow TFX image if the
175   # environment variable 'KUBEFLOW_TFX_IMAGE' is defined. Currently, the tfx
176   # cli tool exports the environment variable to pass to the pipelines.
177   tfx_image = os.environ.get('KUBEFLOW_TFX_IMAGE', None)
178 
179   runner_config = kubeflow_dag_runner.KubeflowDagRunnerConfig(
180       kubeflow_metadata_config=metadata_config,
181       # Specify custom docker image to use.
182       tfx_image=tfx_image
183   )
184 
185   kubeflow_dag_runner.KubeflowDagRunner(config=runner_config).run(
186       _create_pipeline())
187
```
--- 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/tfx/examples/custom_components/slack/example/taxi_pipeline_slack_kubeflow.py b/tfx/examples/custom_components/slack/example/taxi_pipeline_slack_kubeflow.py
--- a/tfx/examples/custom_components/slack/example/taxi_pipeline_slack_kubeflow.py
+++ b/tfx/examples/custom_components/slack/example/taxi_pipeline_slack_kubeflow.py
@@ -145,7 +145,7 @@
       slack_channel_id=_slack_channel_id,
       timeout_sec=3600,
   )
- # Checks whether the model passed the validation steps and pushes the model
+  # Checks whether the model passed the validation steps and pushes the model
   # to a file destination if check passed.
   pusher = Pusher(
       model=trainer.outputs['model'],

{"golden_diff": "diff --git a/tfx/examples/custom_components/slack/example/taxi_pipeline_slack_kubeflow.py b/tfx/examples/custom_components/slack/example/taxi_pipeline_slack_kubeflow.py\n--- a/tfx/examples/custom_components/slack/example/taxi_pipeline_slack_kubeflow.py\n+++ b/tfx/examples/custom_components/slack/example/taxi_pipeline_slack_kubeflow.py\n@@ -145,7 +145,7 @@\n       slack_channel_id=_slack_channel_id,\n       timeout_sec=3600,\n   )\n- # Checks whether the model passed the validation steps and pushes the model\n+  # Checks whether the model passed the validation steps and pushes the model\n   # to a file destination if check passed.\n   pusher = Pusher(\n       model=trainer.outputs['model'],\n", "issue": "just a few typos fixed\nThat Python version check can be removed in a few days when Colab drops Python 2 support.\n", "before_files": [{"content": "# Lint as: python2, python3\n# Copyright 2019 Google LLC. All Rights Reserved.\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\"\"\"Example pipeline to demonstrate custom TFX component.\n\nThis example consists of standard TFX components as well as a custom TFX\ncomponent requesting for manual review through Slack.\n\nThis example along with the custom `SlackComponent` will only serve as an\nexample and will not be supported by TFX team.\n\nThis example runs in Kubeflow with Google Cloud services..\n\"\"\"\n\nfrom __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nimport datetime\nimport os\n\nfrom slack_component.component import SlackComponent\n\nfrom tfx.components import CsvExampleGen\nfrom tfx.components import Evaluator\nfrom tfx.components import ExampleValidator\nfrom tfx.components import ModelValidator\nfrom tfx.components import Pusher\nfrom tfx.components import SchemaGen\nfrom tfx.components import StatisticsGen\nfrom tfx.components import Trainer\nfrom tfx.components import Transform\nfrom tfx.orchestration import pipeline\nfrom tfx.orchestration.kubeflow import kubeflow_dag_runner\nfrom tfx.proto import evaluator_pb2\nfrom tfx.proto import pusher_pb2\nfrom tfx.proto import trainer_pb2\nfrom tfx.utils.dsl_utils import csv_input\n\n# This example assumes that the taxi data is stored in _input_bucket/data/simple\n# and the taxi utility function is in example/taxi_utils_slack.py.\n# Feel free to customize this as needed.\n_input_bucket = 'gs://my-bucket'\n_output_bucket = 'gs://my-bucket'\n_taxi_root = __file__\n_data_root = os.path.join(_input_bucket, 'data', 'simple')\n\n\n# Python module file to inject customized logic into the TFX components. The\n# Transform and Trainer both require user-defined functions to run successfully.\n_taxi_trainer_func = 'example.taxi_utils_slack.trainer_fn'\n_taxi_transformer_func = 'example.taxi_utils_slack.preprocessing_fn'\n# Path which can be listened to by the model server.  Pusher will output the\n# trained model here.\n_serving_model_dir = os.path.join(_taxi_root, 'serving_model/taxi_slack')\n# Slack channel to push the model notifications to.\n_slack_channel_id = 'my-channel-id'\n# Slack token to set up connection.\n_slack_token = os.environ['SLACK_BOT_TOKEN']\n\n# Directory and data locations.  This example assumes all of the chicago taxi\n# example code and metadata library is relative to $HOME, but you can store\n# these files anywhere on your local filesystem.\n_tfx_root = os.path.join(os.environ['HOME'], '/tfx')\n_pipeline_name = 'chicago_taxi_slack_kubeflow'\n_pipeline_root = os.path.join(_input_bucket, _pipeline_name)\n_log_root = os.path.join(_tfx_root, 'logs')\n\n# Airflow-specific configs; these will be passed directly to airflow\n_airflow_config = {\n    'schedule_interval': None,\n    'start_date': datetime.datetime(2019, 1, 1),\n}\n\n\ndef _create_pipeline():\n  \"\"\"Implements the chicago taxi pipeline with TFX.\"\"\"\n  examples = csv_input(_data_root)\n\n  # Brings data into the pipeline or otherwise joins/converts training data.\n  example_gen = CsvExampleGen(input=examples)\n\n  # Computes statistics over data for visualization and example validation.\n  statistics_gen = StatisticsGen(examples=example_gen.outputs['examples'])\n\n  # Generates schema based on statistics files.\n  infer_schema = SchemaGen(statistics=statistics_gen.outputs['statistics'])\n\n  # Performs anomaly detection based on statistics and data schema.\n  validate_stats = ExampleValidator(\n      statistics=statistics_gen.outputs['statistics'],\n      schema=infer_schema.outputs['schema'])\n\n  # Performs transformations and feature engineering in training and serving.\n  transform = Transform(\n      examples=example_gen.outputs['examples'],\n      schema=infer_schema.outputs['schema'],\n      preprocessing_fn=_taxi_transformer_func)\n\n  # Uses user-provided Python function that implements a model using TF-Learn.\n  trainer = Trainer(\n      trainer_fn=_taxi_trainer_func,\n      examples=transform.outputs['transformed_examples'],\n      schema=infer_schema.outputs['schema'],\n      transform_graph=transform.outputs['transform_graph'],\n      train_args=trainer_pb2.TrainArgs(num_steps=10000),\n      eval_args=trainer_pb2.EvalArgs(num_steps=5000))\n\n  # Uses TFMA to compute a evaluation statistics over features of a model.\n  model_analyzer = Evaluator(\n      examples=example_gen.outputs['examples'],\n      model_exports=trainer.outputs['model'],\n      feature_slicing_spec=evaluator_pb2.FeatureSlicingSpec(specs=[\n          evaluator_pb2.SingleSlicingSpec(\n              column_for_slicing=['trip_start_hour'])\n      ]))\n\n  # Performs quality validation of a candidate model (compared to a baseline).\n  model_validator = ModelValidator(\n      examples=example_gen.outputs['examples'], model=trainer.outputs['model'])\n\n  # This custom component serves as a bridge between pipeline and human model\n  # reviewers to enable review-and-push workflow in model development cycle. It\n  # utilizes Slack API to send message to user-defined Slack channel with model\n  # URI info and wait for go / no-go decision from the same Slack channel:\n  #   * To approve the model, users need to reply the thread sent out by the bot\n  #     started by SlackComponent with 'lgtm' or 'approve'.\n  #   * To reject the model, users need to reply the thread sent out by the bot\n  #     started by SlackComponent with 'decline' or 'reject'.\n  slack_validator = SlackComponent(\n      model=trainer.outputs['model'],\n      model_blessing=model_validator.outputs['blessing'],\n      slack_token=_slack_token,\n      slack_channel_id=_slack_channel_id,\n      timeout_sec=3600,\n  )\n # Checks whether the model passed the validation steps and pushes the model\n  # to a file destination if check passed.\n  pusher = Pusher(\n      model=trainer.outputs['model'],\n      model_blessing=slack_validator.outputs['slack_blessing'],\n      push_destination=pusher_pb2.PushDestination(\n          filesystem=pusher_pb2.PushDestination.Filesystem(\n              base_directory=_serving_model_dir)))\n\n  return pipeline.Pipeline(\n      pipeline_name=_pipeline_name,\n      pipeline_root=_pipeline_root,\n      components=[\n          example_gen, statistics_gen, infer_schema, validate_stats, transform,\n          trainer, model_analyzer, model_validator, slack_validator, pusher\n      ],\n      enable_cache=True,\n  )\n\n\nif __name__ == '__main__':\n  # Metadata config. The defaults works work with the installation of\n  # KF Pipelines using Kubeflow. If installing KF Pipelines using the\n  # lightweight deployment option, you may need to override the defaults.\n  metadata_config = kubeflow_dag_runner.get_default_kubeflow_metadata_config()\n\n  # This pipeline automatically injects the Kubeflow TFX image if the\n  # environment variable 'KUBEFLOW_TFX_IMAGE' is defined. Currently, the tfx\n  # cli tool exports the environment variable to pass to the pipelines.\n  tfx_image = os.environ.get('KUBEFLOW_TFX_IMAGE', None)\n\n  runner_config = kubeflow_dag_runner.KubeflowDagRunnerConfig(\n      kubeflow_metadata_config=metadata_config,\n      # Specify custom docker image to use.\n      tfx_image=tfx_image\n  )\n\n  kubeflow_dag_runner.KubeflowDagRunner(config=runner_config).run(\n      _create_pipeline())\n", "path": "tfx/examples/custom_components/slack/example/taxi_pipeline_slack_kubeflow.py"}], "after_files": [{"content": "# Lint as: python2, python3\n# Copyright 2019 Google LLC. All Rights Reserved.\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\"\"\"Example pipeline to demonstrate custom TFX component.\n\nThis example consists of standard TFX components as well as a custom TFX\ncomponent requesting for manual review through Slack.\n\nThis example along with the custom `SlackComponent` will only serve as an\nexample and will not be supported by TFX team.\n\nThis example runs in Kubeflow with Google Cloud services..\n\"\"\"\n\nfrom __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nimport datetime\nimport os\n\nfrom slack_component.component import SlackComponent\n\nfrom tfx.components import CsvExampleGen\nfrom tfx.components import Evaluator\nfrom tfx.components import ExampleValidator\nfrom tfx.components import ModelValidator\nfrom tfx.components import Pusher\nfrom tfx.components import SchemaGen\nfrom tfx.components import StatisticsGen\nfrom tfx.components import Trainer\nfrom tfx.components import Transform\nfrom tfx.orchestration import pipeline\nfrom tfx.orchestration.kubeflow import kubeflow_dag_runner\nfrom tfx.proto import evaluator_pb2\nfrom tfx.proto import pusher_pb2\nfrom tfx.proto import trainer_pb2\nfrom tfx.utils.dsl_utils import csv_input\n\n# This example assumes that the taxi data is stored in _input_bucket/data/simple\n# and the taxi utility function is in example/taxi_utils_slack.py.\n# Feel free to customize this as needed.\n_input_bucket = 'gs://my-bucket'\n_output_bucket = 'gs://my-bucket'\n_taxi_root = __file__\n_data_root = os.path.join(_input_bucket, 'data', 'simple')\n\n\n# Python module file to inject customized logic into the TFX components. The\n# Transform and Trainer both require user-defined functions to run successfully.\n_taxi_trainer_func = 'example.taxi_utils_slack.trainer_fn'\n_taxi_transformer_func = 'example.taxi_utils_slack.preprocessing_fn'\n# Path which can be listened to by the model server.  Pusher will output the\n# trained model here.\n_serving_model_dir = os.path.join(_taxi_root, 'serving_model/taxi_slack')\n# Slack channel to push the model notifications to.\n_slack_channel_id = 'my-channel-id'\n# Slack token to set up connection.\n_slack_token = os.environ['SLACK_BOT_TOKEN']\n\n# Directory and data locations.  This example assumes all of the chicago taxi\n# example code and metadata library is relative to $HOME, but you can store\n# these files anywhere on your local filesystem.\n_tfx_root = os.path.join(os.environ['HOME'], '/tfx')\n_pipeline_name = 'chicago_taxi_slack_kubeflow'\n_pipeline_root = os.path.join(_input_bucket, _pipeline_name)\n_log_root = os.path.join(_tfx_root, 'logs')\n\n# Airflow-specific configs; these will be passed directly to airflow\n_airflow_config = {\n    'schedule_interval': None,\n    'start_date': datetime.datetime(2019, 1, 1),\n}\n\n\ndef _create_pipeline():\n  \"\"\"Implements the chicago taxi pipeline with TFX.\"\"\"\n  examples = csv_input(_data_root)\n\n  # Brings data into the pipeline or otherwise joins/converts training data.\n  example_gen = CsvExampleGen(input=examples)\n\n  # Computes statistics over data for visualization and example validation.\n  statistics_gen = StatisticsGen(examples=example_gen.outputs['examples'])\n\n  # Generates schema based on statistics files.\n  infer_schema = SchemaGen(statistics=statistics_gen.outputs['statistics'])\n\n  # Performs anomaly detection based on statistics and data schema.\n  validate_stats = ExampleValidator(\n      statistics=statistics_gen.outputs['statistics'],\n      schema=infer_schema.outputs['schema'])\n\n  # Performs transformations and feature engineering in training and serving.\n  transform = Transform(\n      examples=example_gen.outputs['examples'],\n      schema=infer_schema.outputs['schema'],\n      preprocessing_fn=_taxi_transformer_func)\n\n  # Uses user-provided Python function that implements a model using TF-Learn.\n  trainer = Trainer(\n      trainer_fn=_taxi_trainer_func,\n      examples=transform.outputs['transformed_examples'],\n      schema=infer_schema.outputs['schema'],\n      transform_graph=transform.outputs['transform_graph'],\n      train_args=trainer_pb2.TrainArgs(num_steps=10000),\n      eval_args=trainer_pb2.EvalArgs(num_steps=5000))\n\n  # Uses TFMA to compute a evaluation statistics over features of a model.\n  model_analyzer = Evaluator(\n      examples=example_gen.outputs['examples'],\n      model_exports=trainer.outputs['model'],\n      feature_slicing_spec=evaluator_pb2.FeatureSlicingSpec(specs=[\n          evaluator_pb2.SingleSlicingSpec(\n              column_for_slicing=['trip_start_hour'])\n      ]))\n\n  # Performs quality validation of a candidate model (compared to a baseline).\n  model_validator = ModelValidator(\n      examples=example_gen.outputs['examples'], model=trainer.outputs['model'])\n\n  # This custom component serves as a bridge between pipeline and human model\n  # reviewers to enable review-and-push workflow in model development cycle. It\n  # utilizes Slack API to send message to user-defined Slack channel with model\n  # URI info and wait for go / no-go decision from the same Slack channel:\n  #   * To approve the model, users need to reply the thread sent out by the bot\n  #     started by SlackComponent with 'lgtm' or 'approve'.\n  #   * To reject the model, users need to reply the thread sent out by the bot\n  #     started by SlackComponent with 'decline' or 'reject'.\n  slack_validator = SlackComponent(\n      model=trainer.outputs['model'],\n      model_blessing=model_validator.outputs['blessing'],\n      slack_token=_slack_token,\n      slack_channel_id=_slack_channel_id,\n      timeout_sec=3600,\n  )\n  # Checks whether the model passed the validation steps and pushes the model\n  # to a file destination if check passed.\n  pusher = Pusher(\n      model=trainer.outputs['model'],\n      model_blessing=slack_validator.outputs['slack_blessing'],\n      push_destination=pusher_pb2.PushDestination(\n          filesystem=pusher_pb2.PushDestination.Filesystem(\n              base_directory=_serving_model_dir)))\n\n  return pipeline.Pipeline(\n      pipeline_name=_pipeline_name,\n      pipeline_root=_pipeline_root,\n      components=[\n          example_gen, statistics_gen, infer_schema, validate_stats, transform,\n          trainer, model_analyzer, model_validator, slack_validator, pusher\n      ],\n      enable_cache=True,\n  )\n\n\nif __name__ == '__main__':\n  # Metadata config. The defaults works work with the installation of\n  # KF Pipelines using Kubeflow. If installing KF Pipelines using the\n  # lightweight deployment option, you may need to override the defaults.\n  metadata_config = kubeflow_dag_runner.get_default_kubeflow_metadata_config()\n\n  # This pipeline automatically injects the Kubeflow TFX image if the\n  # environment variable 'KUBEFLOW_TFX_IMAGE' is defined. Currently, the tfx\n  # cli tool exports the environment variable to pass to the pipelines.\n  tfx_image = os.environ.get('KUBEFLOW_TFX_IMAGE', None)\n\n  runner_config = kubeflow_dag_runner.KubeflowDagRunnerConfig(\n      kubeflow_metadata_config=metadata_config,\n      # Specify custom docker image to use.\n      tfx_image=tfx_image\n  )\n\n  kubeflow_dag_runner.KubeflowDagRunner(config=runner_config).run(\n      _create_pipeline())\n", "path": "tfx/examples/custom_components/slack/example/taxi_pipeline_slack_kubeflow.py"}]}