Allow Pipeline objects to be used in Python with statements

----Release Notes----
One can now write

  with df.Pipeline(...) as p:
    p | df.io.Read(...) | Map(...) | ...

and p.run() will be called automatically at the end of the block.

[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=124308722

Author: robertwb

google/cloud/dataflow/pipeline.py

@@ -157,6 +157,13 @@ def run(self):
         shutil.rmtree(tmpdir)
     return self.runner.run(self)
 
+  def __enter__(self):
+    return self
+
+  def __exit__(self, exc_type, exc_val, exc_tb):
+    if not exc_type:
+      self.run()
+
   def visit(self, visitor):
     """Visits depth-first every node of a pipeline's DAG.
 

google/cloud/dataflow/pipeline_test.py

@@ -245,6 +245,14 @@ def create_dupes(o, _):
             ('oom:combine/GroupByKey/group_by_window', None): 1,
             ('oom:combine/Combine/ParDo(CombineValuesDoFn)', None): 1})
 
+  def test_pipeline_as_context(self):
+    def raise_exception(exn):
+      raise exn
+    with self.assertRaises(ValueError):
+      with Pipeline(self.runner_name) as p:
+        # pylint: disable=expression-not-assigned
+        p | Create([ValueError]) | Map(raise_exception)
+
   def test_eager_pipeline(self):
     p = Pipeline('EagerPipelineRunner')
     self.assertEqual([1, 4, 9], p | Create([1, 2, 3]) | Map(lambda x: x*x))