beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dhalp...@apache.org
Subject [1/2] incubator-beam git commit: Enables more linting rules.
Date Wed, 29 Jun 2016 01:23:33 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/python-sdk cb1842c5b -> 887bef19d


Enables more linting rules.

- Fixes some import related warnings and enabled related pylint rules.
- Use pep8 for blank line related style check. pylint does not check
  for pep8 related to pep8. And pep8 is not configurable for indentations.
  So we need both tools.
- Fixed existing lint error related to blank lines.


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/57a4495d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/57a4495d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/57a4495d

Branch: refs/heads/python-sdk
Commit: 57a4495d996acdd4eb69f7219993c9c5d992b2b1
Parents: cb1842c
Author: Ahmet Altay <altay@google.com>
Authored: Mon Jun 27 17:59:35 2016 -0700
Committer: Dan Halperin <dhalperi@google.com>
Committed: Tue Jun 28 18:23:10 2016 -0700

----------------------------------------------------------------------
 sdks/python/.pylintrc                           |  3 -
 sdks/python/apache_beam/__init__.py             |  3 +-
 sdks/python/apache_beam/coders/coder_impl.py    |  4 +-
 sdks/python/apache_beam/coders/coders.py        |  4 +-
 .../apache_beam/coders/fast_coders_test.py      |  2 +-
 .../apache_beam/coders/slow_coders_test.py      |  2 +-
 sdks/python/apache_beam/coders/stream_test.py   |  4 +-
 .../complete/juliaset/juliaset/juliaset.py      |  9 ++-
 .../examples/cookbook/bigquery_schema.py        |  2 +-
 .../apache_beam/examples/snippets/snippets.py   | 19 ++++---
 .../examples/snippets/snippets_test.py          | 58 +++++++++++++-------
 sdks/python/apache_beam/internal/apiclient.py   |  5 +-
 .../apache_beam/internal/json_value_test.py     |  1 +
 sdks/python/apache_beam/internal/pickler.py     |  2 +-
 sdks/python/apache_beam/internal/util.py        |  1 +
 sdks/python/apache_beam/io/avroio.py            |  3 +-
 sdks/python/apache_beam/io/bigquery.py          |  8 +--
 sdks/python/apache_beam/io/bigquery_test.py     |  6 +-
 sdks/python/apache_beam/io/fileio.py            | 16 +++---
 sdks/python/apache_beam/io/gcsio.py             |  7 ++-
 sdks/python/apache_beam/io/gcsio_test.py        |  5 +-
 sdks/python/apache_beam/pvalue.py               |  8 +--
 sdks/python/apache_beam/runners/common.py       |  6 ++
 .../apache_beam/runners/dataflow_runner.py      |  5 +-
 .../python/apache_beam/runners/direct_runner.py |  2 +-
 sdks/python/apache_beam/runners/runner.py       |  6 +-
 .../python/apache_beam/transforms/aggregator.py |  2 +-
 sdks/python/apache_beam/transforms/combiners.py | 10 ++--
 .../apache_beam/transforms/combiners_test.py    |  4 ++
 sdks/python/apache_beam/transforms/core.py      | 21 +++----
 .../apache_beam/transforms/cy_combiners.py      | 53 ++++++++++++++++++
 .../python/apache_beam/transforms/ptransform.py | 13 +++--
 .../apache_beam/transforms/ptransform_test.py   | 10 ++++
 .../python/apache_beam/transforms/sideinputs.py |  9 ++-
 sdks/python/apache_beam/transforms/trigger.py   |  1 +
 .../apache_beam/transforms/trigger_test.py      |  5 +-
 sdks/python/apache_beam/transforms/util.py      |  1 +
 .../apache_beam/transforms/window_test.py       |  1 +
 .../apache_beam/typehints/typehints_test.py     |  3 +
 .../python/apache_beam/utils/dependency_test.py |  1 +
 sdks/python/apache_beam/utils/retry_test.py     |  4 +-
 sdks/python/run_pylint.sh                       |  7 ++-
 sdks/python/tox.ini                             | 12 +++-
 43 files changed, 234 insertions(+), 114 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/.pylintrc
----------------------------------------------------------------------
diff --git a/sdks/python/.pylintrc b/sdks/python/.pylintrc
index edf13ee..efe8147 100644
--- a/sdks/python/.pylintrc
+++ b/sdks/python/.pylintrc
@@ -83,7 +83,6 @@ disable =
   arguments-differ,
   attribute-defined-outside-init,
   bad-builtin,
-  bad-option-value,
   bad-super-call,
   broad-except,
   consider-using-enumerate,
@@ -131,8 +130,6 @@ disable =
   unused-wildcard-import,
   used-before-assignment,
   wildcard-import,
-  wrong-import-order,
-  wrong-import-position,
 
 
 [REPORTS]

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/__init__.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/__init__.py b/sdks/python/apache_beam/__init__.py
index c9bc736..eed251b 100644
--- a/sdks/python/apache_beam/__init__.py
+++ b/sdks/python/apache_beam/__init__.py
@@ -71,7 +71,7 @@ if sys.version_info.major != 2:
       'Dataflow SDK for Python is supported only on Python 2.7. '
       'It is not supported on Python [%s].' % sys.version)
 
-
+# pylint: disable=wrong-import-position
 import apache_beam.internal.pickler
 
 from apache_beam import coders
@@ -79,3 +79,4 @@ from apache_beam import io
 from apache_beam import typehints
 from apache_beam.pipeline import Pipeline
 from apache_beam.transforms import *
+# pylint: enable=wrong-import-position

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/coders/coder_impl.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/coders/coder_impl.py b/sdks/python/apache_beam/coders/coder_impl.py
index 07b6711..a623f2b 100644
--- a/sdks/python/apache_beam/coders/coder_impl.py
+++ b/sdks/python/apache_beam/coders/coder_impl.py
@@ -29,7 +29,7 @@ import collections
 from cPickle import loads, dumps
 
 
-# pylint: disable=g-import-not-at-top
+# pylint: disable=wrong-import-order, wrong-import-position
 try:
   # Don't depend on the full dataflow sdk to test coders.
   from apache_beam.transforms.window import WindowedValue
@@ -43,7 +43,7 @@ try:
 except ImportError:
   from slow_stream import InputStream as create_InputStream
   from slow_stream import OutputStream as create_OutputStream
-# pylint: enable=g-import-not-at-top
+# pylint: enable=wrong-import-order, wrong-import-position
 
 
 class CoderImpl(object):

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/coders/coders.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/coders/coders.py b/sdks/python/apache_beam/coders/coders.py
index 6d5b10a..619586f 100644
--- a/sdks/python/apache_beam/coders/coders.py
+++ b/sdks/python/apache_beam/coders/coders.py
@@ -24,7 +24,7 @@ import cPickle as pickle
 from apache_beam.coders import coder_impl
 
 
-# pylint: disable=g-import-not-at-top
+# pylint: disable=wrong-import-order, wrong-import-position
 # Avoid dependencies on the full SDK.
 try:
   # Import dill from the pickler module to make sure our monkey-patching of dill
@@ -46,7 +46,7 @@ def serialize_coder(coder):
 def deserialize_coder(serialized):
   from apache_beam.internal import pickler
   return pickler.loads(serialized.split('$', 1)[1])
-# pylint: enable=g-import-not-at-top
+# pylint: enable=wrong-import-order, wrong-import-position
 
 
 class Coder(object):

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/coders/fast_coders_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/coders/fast_coders_test.py b/sdks/python/apache_beam/coders/fast_coders_test.py
index 466fd57..55cf16c 100644
--- a/sdks/python/apache_beam/coders/fast_coders_test.py
+++ b/sdks/python/apache_beam/coders/fast_coders_test.py
@@ -28,7 +28,7 @@ from apache_beam.coders.coders_test_common import *
 class FastCoders(unittest.TestCase):
 
   def test_using_fast_impl(self):
-    # pylint: disable=g-import-not-at-top
+    # pylint: disable=wrong-import-order, wrong-import-position
     # pylint: disable=unused-variable
     import apache_beam.coders.stream
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/coders/slow_coders_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/coders/slow_coders_test.py b/sdks/python/apache_beam/coders/slow_coders_test.py
index 26e6fa7..62149a3 100644
--- a/sdks/python/apache_beam/coders/slow_coders_test.py
+++ b/sdks/python/apache_beam/coders/slow_coders_test.py
@@ -30,7 +30,7 @@ class SlowCoders(unittest.TestCase):
   def test_using_slow_impl(self):
     # Assert that we are not using the compiled implementation.
     with self.assertRaises(ImportError):
-      # pylint: disable=g-import-not-at-top
+      # pylint: disable=wrong-import-order, wrong-import-position
       # pylint: disable=unused-variable
       import apache_beam.coders.stream
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/coders/stream_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/coders/stream_test.py b/sdks/python/apache_beam/coders/stream_test.py
index f2163ca..cfd627f 100644
--- a/sdks/python/apache_beam/coders/stream_test.py
+++ b/sdks/python/apache_beam/coders/stream_test.py
@@ -139,7 +139,7 @@ class StreamTest(unittest.TestCase):
 
 
 try:
-  # pylint: disable=g-import-not-at-top
+  # pylint: disable=wrong-import-position
   from apache_beam.coders import stream
 
   class FastStreamTest(StreamTest):
@@ -148,14 +148,12 @@ try:
     OutputStream = stream.OutputStream
     ByteCountingOutputStream = stream.ByteCountingOutputStream
 
-
   class SlowFastStreamTest(StreamTest):
     """Runs the test with compiled and uncompiled stream classes."""
     InputStream = stream.InputStream
     OutputStream = slow_stream.OutputStream
     ByteCountingOutputStream = slow_stream.ByteCountingOutputStream
 
-
   class FastSlowStreamTest(StreamTest):
     """Runs the test with uncompiled and compiled stream classes."""
     InputStream = slow_stream.InputStream

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/examples/complete/juliaset/juliaset/juliaset.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/examples/complete/juliaset/juliaset/juliaset.py b/sdks/python/apache_beam/examples/complete/juliaset/juliaset/juliaset.py
index 1625418..2bc37e6 100644
--- a/sdks/python/apache_beam/examples/complete/juliaset/juliaset/juliaset.py
+++ b/sdks/python/apache_beam/examples/complete/juliaset/juliaset/juliaset.py
@@ -51,14 +51,15 @@ def generate_julia_set_colors(pipeline, c, n, max_iterations):
 
   julia_set_colors = (pipeline
                       | beam.Create('add points', point_set(n))
-                      | beam.Map(get_julia_set_point_color, c, n, max_iterations))
+                      | beam.Map(
+                          get_julia_set_point_color, c, n, max_iterations))
 
   return julia_set_colors
 
 
 def generate_julia_set_visualization(data, n, max_iterations):
   """Generate the pixel matrix for rendering the julia set as an image."""
-  import numpy as np  # pylint: disable=g-import-not-at-top
+  import numpy as np  # pylint: disable=wrong-import-order, wrong-import-position
   colors = []
   for r in range(0, 256, 16):
     for g in range(0, 256, 16):
@@ -74,7 +75,7 @@ def generate_julia_set_visualization(data, n, max_iterations):
 
 def save_julia_set_visualization(out_file, image_array):
   """Save the fractal image of our julia set as a png."""
-  from matplotlib import pyplot as plt  # pylint: disable=g-import-not-at-top
+  from matplotlib import pyplot as plt  # pylint: disable=wrong-import-order, wrong-import-position
   plt.imsave(out_file, image_array, format='png')
 
 
@@ -104,13 +105,11 @@ def run(argv=None):  # pylint: disable=missing-docstring
   # Group each coordinate triplet by its x value, then write the coordinates to
   # the output file with an x-coordinate grouping per line.
   # pylint: disable=expression-not-assigned
-  # pylint: disable=g-long-lambda
   (coordinates | beam.Map('x coord key', lambda (x, y, i): (x, (x, y, i)))
    | beam.GroupByKey('x coord') | beam.Map(
        'format',
        lambda (k, coords): ' '.join('(%s, %s, %s)' % coord for coord in coords))
    | beam.io.Write('write', beam.io.TextFileSink(known_args.coordinate_output)))
-  # pylint: enable=g-long-lambda
   # pylint: enable=expression-not-assigned
   p.run()
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/examples/cookbook/bigquery_schema.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/examples/cookbook/bigquery_schema.py b/sdks/python/apache_beam/examples/cookbook/bigquery_schema.py
index 99a967c..7c420fb 100644
--- a/sdks/python/apache_beam/examples/cookbook/bigquery_schema.py
+++ b/sdks/python/apache_beam/examples/cookbook/bigquery_schema.py
@@ -44,7 +44,7 @@ def run(argv=None):
 
   p = beam.Pipeline(argv=pipeline_args)
 
-  from apache_beam.internal.clients import bigquery  # pylint: disable=g-import-not-at-top
+  from apache_beam.internal.clients import bigquery  # pylint: disable=wrong-import-order, wrong-import-position
 
   table_schema = bigquery.TableSchema()
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/examples/snippets/snippets.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/examples/snippets/snippets.py b/sdks/python/apache_beam/examples/snippets/snippets.py
index e7af26a..f5bbc66 100644
--- a/sdks/python/apache_beam/examples/snippets/snippets.py
+++ b/sdks/python/apache_beam/examples/snippets/snippets.py
@@ -40,8 +40,7 @@ import apache_beam as beam
 # pylint:disable=expression-not-assigned
 # pylint:disable=redefined-outer-name
 # pylint:disable=unused-variable
-# pylint:disable=g-doc-args
-# pylint:disable=g-import-not-at-top
+# pylint:disable=wrong-import-order, wrong-import-position
 
 
 class SnippetUtils(object):
@@ -95,7 +94,7 @@ def construct_pipeline(renames):
 
   # [START pipelines_constructing_reading]
   lines = p | beam.io.Read('ReadMyFile',
-                      beam.io.TextFileSource('gs://some/inputData.txt'))
+                           beam.io.TextFileSource('gs://some/inputData.txt'))
   # [END pipelines_constructing_reading]
 
   # [START pipelines_constructing_applying]
@@ -106,7 +105,8 @@ def construct_pipeline(renames):
   # [START pipelines_constructing_writing]
   filtered_words = reversed_words | beam.Filter('FilterWords', filter_words)
   filtered_words | beam.io.Write('WriteMyFile',
-                               beam.io.TextFileSink('gs://some/outputData.txt'))
+                                 beam.io.TextFileSink(
+                                     'gs://some/outputData.txt'))
   # [END pipelines_constructing_writing]
 
   p.visit(SnippetUtils.RenameFiles(renames))
@@ -304,7 +304,8 @@ def pipeline_options_command_line(argv):
 
   # Create the Pipeline with remaining arguments.
   p = beam.Pipeline(argv=pipeline_args)
-  lines = p | beam.io.Read('ReadFromText', beam.io.TextFileSource(known_args.input))
+  lines = p | beam.io.Read('ReadFromText',
+                           beam.io.TextFileSource(known_args.input))
   lines | beam.io.Write('WriteToText', beam.io.TextFileSink(known_args.output))
   # [END pipeline_options_command_line]
 
@@ -594,7 +595,8 @@ def examples_wordcount_debugging(renames):
       | beam.ParDo('FilterText', FilterTextFn('Flourish|stomach')))
 
   # [START example_wordcount_debugging_assert]
-  beam.assert_that(filtered_words, beam.equal_to([('Flourish', 3), ('stomach', 1)]))
+  beam.assert_that(
+      filtered_words, beam.equal_to([('Flourish', 3), ('stomach', 1)]))
   # [END example_wordcount_debugging_assert]
 
   output = (filtered_words
@@ -634,7 +636,7 @@ def model_textio(renames):
   # [START model_pipelineio_write]
   filtered_words | beam.io.Write(
       'WriteToText', beam.io.TextFileSink('gs://my_bucket/path/to/numbers',
-                                        file_name_suffix='.csv'))
+                                          file_name_suffix='.csv'))
   # [END model_pipelineio_write]
   # [END model_textio_write]
 
@@ -762,6 +764,7 @@ def model_multiple_pcollections_partition(contents, output_path):
   URL: https://cloud.google.com/dataflow/model/multiple-pcollections
   """
   some_hash_fn = lambda s: ord(s[0])
+
   def get_percentile(i):
     """Assume i in [0,100)."""
     return i
@@ -770,6 +773,7 @@ def model_multiple_pcollections_partition(contents, output_path):
   p = beam.Pipeline(options=PipelineOptions())
 
   students = p | beam.Create(contents)
+
   # [START model_multiple_pcollections_partition]
   def partition_fn(student, num_partitions):
     return int(get_percentile(student) * num_partitions / 100)
@@ -872,4 +876,3 @@ class Count(beam.PTransform):
         | beam.Map('Init', lambda v: (v, 1))
         | beam.CombinePerKey(sum))
 # [END model_library_transforms_count]
-# pylint: enable=g-wrong-blank-lines

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/examples/snippets/snippets_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/examples/snippets/snippets_test.py b/sdks/python/apache_beam/examples/snippets/snippets_test.py
index f2f9552..87ce266 100644
--- a/sdks/python/apache_beam/examples/snippets/snippets_test.py
+++ b/sdks/python/apache_beam/examples/snippets/snippets_test.py
@@ -43,6 +43,7 @@ class ParDoTest(unittest.TestCase):
     # the text of the doc.
 
     words = ['aa', 'bbb', 'c']
+
     # [START model_pardo_pardo]
     class ComputeWordLengthFn(beam.DoFn):
       def process(self, context):
@@ -57,6 +58,7 @@ class ParDoTest(unittest.TestCase):
 
   def test_pardo_yield(self):
     words = ['aa', 'bbb', 'c']
+
     # [START model_pardo_yield]
     class ComputeWordLengthFn(beam.DoFn):
       def process(self, context):
@@ -84,11 +86,12 @@ class ParDoTest(unittest.TestCase):
 
   def test_pardo_using_flatmap_yield(self):
     words = ['aA', 'bbb', 'C']
+
     # [START model_pardo_using_flatmap_yield]
     def capitals(word):
       for letter in word:
         if 'A' <= letter <= 'Z':
-            yield letter
+          yield letter
     all_capitals = words | beam.FlatMap(capitals)
     # [END model_pardo_using_flatmap_yield]
 
@@ -113,27 +116,31 @@ class ParDoTest(unittest.TestCase):
         yield word
 
     # Construct a deferred side input.
-    avg_word_len = words | beam.Map(len) | beam.CombineGlobally(beam.combiners.MeanCombineFn())
+    avg_word_len = (words
+                    | beam.Map(len)
+                    | beam.CombineGlobally(beam.combiners.MeanCombineFn()))
 
     # Call with explicit side inputs.
     small_words = words | beam.FlatMap('small', filter_using_length, 0, 3)
 
     # A single deferred side input.
-    larger_than_average = words | beam.FlatMap('large',
-                                             filter_using_length,
-                                             lower_bound=pvalue.AsSingleton(avg_word_len))
+    larger_than_average = (words
+                           | beam.FlatMap('large', filter_using_length,
+                                          lower_bound=pvalue.AsSingleton(
+                                              avg_word_len)))
 
     # Mix and match.
     small_but_nontrivial = words | beam.FlatMap(filter_using_length,
-                                              lower_bound=2,
-                                              upper_bound=pvalue.AsSingleton(avg_word_len))
+                                                lower_bound=2,
+                                                upper_bound=pvalue.AsSingleton(
+                                                    avg_word_len))
     # [END model_pardo_side_input]
 
     beam.assert_that(small_words, beam.equal_to(['a', 'bb', 'ccc']))
     beam.assert_that(larger_than_average, beam.equal_to(['ccc', 'dddd']),
-                   label='larger_than_average')
+                     label='larger_than_average')
     beam.assert_that(small_but_nontrivial, beam.equal_to(['bb']),
-                   label='small_but_not_trivial')
+                     label='small_but_not_trivial')
     p.run()
 
   def test_pardo_side_input_dofn(self):
@@ -170,9 +177,8 @@ class ParDoTest(unittest.TestCase):
 
     # [START model_pardo_with_side_outputs]
     results = (words | beam.ParDo(ProcessWords(), cutoff_length=2, marker='x')
-                         .with_outputs('above_cutoff_lengths',
-                                       'marked strings',
-                                       main='below_cutoff_strings'))
+               .with_outputs('above_cutoff_lengths', 'marked strings',
+                             main='below_cutoff_strings'))
     below = results.below_cutoff_strings
     above = results.above_cutoff_lengths
     marked = results['marked strings']  # indexing works as well
@@ -183,10 +189,12 @@ class ParDoTest(unittest.TestCase):
     self.assertEqual({'xyz'}, set(marked))
 
     # [START model_pardo_with_side_outputs_iter]
-    below, above, marked = (words | beam.ParDo(ProcessWords(), cutoff_length=2, marker='x')
-                                      .with_outputs('above_cutoff_lengths',
-                                                    'marked strings',
-                                                    main='below_cutoff_strings'))
+    below, above, marked = (words
+                            | beam.ParDo(
+                                ProcessWords(), cutoff_length=2, marker='x')
+                            .with_outputs('above_cutoff_lengths',
+                                          'marked strings',
+                                          main='below_cutoff_strings'))
     # [END model_pardo_with_side_outputs_iter]
 
     self.assertEqual({'a', 'an'}, set(below))
@@ -195,6 +203,7 @@ class ParDoTest(unittest.TestCase):
 
   def test_pardo_with_undeclared_side_outputs(self):
     numbers = [1, 2, 3, 4, 5, 10, 20]
+
     # [START model_pardo_with_side_outputs_undeclared]
     def even_odd(x):
       yield pvalue.SideOutputValue('odd' if x % 2 else 'even', x)
@@ -258,6 +267,7 @@ class TypeHintsTest(unittest.TestCase):
     # Helps document the contract and checks it at pipeline construction time.
     # [START type_hints_transform]
     T = beam.typehints.TypeVariable('T')
+
     @beam.typehints.with_input_types(T)
     @beam.typehints.with_output_types(beam.typehints.Tuple[int, T])
     class MyTransform(beam.PTransform):
@@ -316,7 +326,8 @@ class TypeHintsTest(unittest.TestCase):
     totals = (
         lines
         | beam.Map(parse_player_and_score)
-        | beam.CombinePerKey(sum).with_input_types(beam.typehints.Tuple[Player, int]))
+        | beam.CombinePerKey(sum).with_input_types(
+            beam.typehints.Tuple[Player, int]))
     # [END type_hints_deterministic_key]
 
     self.assertEquals(
@@ -491,12 +502,15 @@ class CombineTest(unittest.TestCase):
         ('cat', 1), ('cat', 5), ('cat', 9), ('cat', 1),
         ('dog', 5), ('dog', 2)]
     # [START combine_per_key]
-    avg_accuracy_per_player = player_accuracies | beam.CombinePerKey(beam.combiners.MeanCombineFn())
+    avg_accuracy_per_player = (player_accuracies
+                               | beam.CombinePerKey(
+                                   beam.combiners.MeanCombineFn()))
     # [END combine_per_key]
     self.assertEqual({('cat', 4.0), ('dog', 3.5)}, set(avg_accuracy_per_player))
 
   def test_combine_concat(self):
     pc = ['a', 'b']
+
     # [START combine_concat]
     def concat(values, separator=', '):
       return separator.join(values)
@@ -511,6 +525,7 @@ class CombineTest(unittest.TestCase):
   def test_bounded_sum(self):
     # [START combine_bounded_sum]
     pc = [1, 10, 100, 1000]
+
     def bounded_sum(values, bound=500):
       return min(sum(values), bound)
     small_sum = pc | beam.CombineGlobally(bounded_sum)              # [500]
@@ -524,23 +539,26 @@ class CombineTest(unittest.TestCase):
     # [START combine_reduce]
     import functools
     import operator
-    product = factors | beam.CombineGlobally(functools.partial(reduce, operator.mul), 1)
+    product = factors | beam.CombineGlobally(
+        functools.partial(reduce, operator.mul), 1)
     # [END combine_reduce]
     self.assertEqual([210], product)
 
   def test_custom_average(self):
     pc = [2, 3, 5, 7]
 
-
     # [START combine_custom_average]
     class AverageFn(beam.CombineFn):
       def create_accumulator(self):
         return (0.0, 0)
+
       def add_input(self, (sum, count), input):
         return sum + input, count + 1
+
       def merge_accumulators(self, accumulators):
         sums, counts = zip(*accumulators)
         return sum(sums), sum(counts)
+
       def extract_output(self, (sum, count)):
         return sum / count if count else float('NaN')
     average = pc | beam.CombineGlobally(AverageFn())

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/internal/apiclient.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/internal/apiclient.py b/sdks/python/apache_beam/internal/apiclient.py
index a363511..7dfb035 100644
--- a/sdks/python/apache_beam/internal/apiclient.py
+++ b/sdks/python/apache_beam/internal/apiclient.py
@@ -25,6 +25,9 @@ import re
 import time
 
 
+from apitools.base.py import encoding
+from apitools.base.py import exceptions
+
 from apache_beam import utils
 from apache_beam import version
 from apache_beam.internal import pickler
@@ -40,8 +43,6 @@ from apache_beam.utils.options import GoogleCloudOptions
 from apache_beam.utils.options import StandardOptions
 from apache_beam.utils.options import WorkerOptions
 
-from apitools.base.py import encoding
-from apitools.base.py import exceptions
 from apache_beam.internal.clients import storage
 import apache_beam.internal.clients.dataflow as dataflow
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/internal/json_value_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/internal/json_value_test.py b/sdks/python/apache_beam/internal/json_value_test.py
index 9b26ab2..f2ae0c1 100644
--- a/sdks/python/apache_beam/internal/json_value_test.py
+++ b/sdks/python/apache_beam/internal/json_value_test.py
@@ -24,6 +24,7 @@ from apitools.base.py.extra_types import JsonObject
 from apache_beam.internal.json_value import from_json_value
 from apache_beam.internal.json_value import to_json_value
 
+
 class JsonValueTest(unittest.TestCase):
 
   def test_string_to(self):

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/internal/pickler.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/internal/pickler.py b/sdks/python/apache_beam/internal/pickler.py
index edf63ad..898e04b 100644
--- a/sdks/python/apache_beam/internal/pickler.py
+++ b/sdks/python/apache_beam/internal/pickler.py
@@ -159,13 +159,13 @@ if 'save_module' in dir(dill.dill):
       return old_save_module_dict(pickler, obj)
   dill.dill.save_module_dict = new_save_module_dict
 
-
   def _nest_dill_logging():
     """Prefix all dill logging with its depth in the callstack.
 
     Useful for debugging pickling of deeply nested structures.
     """
     old_log_info = dill.dill.log.info
+
     def new_log_info(msg, *args, **kwargs):
       old_log_info(
           ('1 2 3 4 5 6 7 8 9 0 ' * 10)[:len(traceback.extract_stack())] + msg,

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/internal/util.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/internal/util.py b/sdks/python/apache_beam/internal/util.py
index 50ea2f6..ad60ba6 100644
--- a/sdks/python/apache_beam/internal/util.py
+++ b/sdks/python/apache_beam/internal/util.py
@@ -58,6 +58,7 @@ def remove_objects_from_args(args, kwargs, pvalue_classes):
     a placeholder value.
   """
   pvals = []
+
   def swapper(value):
     pvals.append(value)
     return ArgumentPlaceholder()

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/io/avroio.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/io/avroio.py b/sdks/python/apache_beam/io/avroio.py
index 022a68d..25412af 100644
--- a/sdks/python/apache_beam/io/avroio.py
+++ b/sdks/python/apache_beam/io/avroio.py
@@ -21,11 +21,12 @@ import os
 import StringIO
 import zlib
 
-from apache_beam.io import filebasedsource
 from avro import datafile
 from avro import io as avro_io
 from avro import schema
 
+from apache_beam.io import filebasedsource
+
 
 class AvroSource(filebasedsource.FileBasedSource):
   """A source for reading Avro files.

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/io/bigquery.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/io/bigquery.py b/sdks/python/apache_beam/io/bigquery.py
index 9d33134..f2c56dc 100644
--- a/sdks/python/apache_beam/io/bigquery.py
+++ b/sdks/python/apache_beam/io/bigquery.py
@@ -110,6 +110,8 @@ import re
 import time
 import uuid
 
+from apitools.base.py.exceptions import HttpError
+
 from apache_beam import coders
 from apache_beam.internal import auth
 from apache_beam.internal.json_value import from_json_value
@@ -118,15 +120,13 @@ from apache_beam.io import iobase
 from apache_beam.utils import retry
 from apache_beam.utils.options import GoogleCloudOptions
 
-from apitools.base.py.exceptions import HttpError
-
 # Protect against environments where bigquery library is not available.
-# pylint: disable=g-import-not-at-top
+# pylint: disable=wrong-import-order, wrong-import-position
 try:
   from apache_beam.internal.clients import bigquery
 except ImportError:
   pass
-# pylint: enable=g-import-not-at-top
+# pylint: enable=wrong-import-order, wrong-import-position
 
 
 __all__ = [

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/io/bigquery_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/io/bigquery_test.py b/sdks/python/apache_beam/io/bigquery_test.py
index 7c31526..2bca0dc 100644
--- a/sdks/python/apache_beam/io/bigquery_test.py
+++ b/sdks/python/apache_beam/io/bigquery_test.py
@@ -22,16 +22,16 @@ import logging
 import time
 import unittest
 
+from apitools.base.py.exceptions import HttpError
 import mock
+
 import apache_beam as beam
+from apache_beam.internal.clients import bigquery
 from apache_beam.internal.json_value import to_json_value
 from apache_beam.io.bigquery import RowAsDictJsonCoder
 from apache_beam.io.bigquery import TableRowJsonCoder
 from apache_beam.utils.options import PipelineOptions
 
-from apitools.base.py.exceptions import HttpError
-from apache_beam.internal.clients import bigquery
-
 
 class TestRowAsDictJsonCoder(unittest.TestCase):
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/io/fileio.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/io/fileio.py b/sdks/python/apache_beam/io/fileio.py
index 560d935..115bc0e 100644
--- a/sdks/python/apache_beam/io/fileio.py
+++ b/sdks/python/apache_beam/io/fileio.py
@@ -172,7 +172,7 @@ class ChannelFactory(object):
   @staticmethod
   def open(path, mode, mime_type):
     if path.startswith('gs://'):
-      # pylint: disable=g-import-not-at-top
+      # pylint: disable=wrong-import-order, wrong-import-position
       from apache_beam.io import gcsio
       return gcsio.GcsIO().open(path, mode, mime_type=mime_type)
     else:
@@ -182,7 +182,7 @@ class ChannelFactory(object):
   def rename(src, dst):
     if src.startswith('gs://'):
       assert dst.startswith('gs://'), dst
-      # pylint: disable=g-import-not-at-top
+      # pylint: disable=wrong-import-order, wrong-import-position
       from apache_beam.io import gcsio
       gcsio.GcsIO().rename(src, dst)
     else:
@@ -197,7 +197,7 @@ class ChannelFactory(object):
       assert dst.startswith('gs://'), dst
       assert src.endswith('/'), src
       assert dst.endswith('/'), dst
-      # pylint: disable=g-import-not-at-top
+      # pylint: disable=wrong-import-order, wrong-import-position
       from apache_beam.io import gcsio
       gcsio.GcsIO().copytree(src, dst)
     else:
@@ -211,7 +211,7 @@ class ChannelFactory(object):
   @staticmethod
   def exists(path):
     if path.startswith('gs://'):
-      # pylint: disable=g-import-not-at-top
+      # pylint: disable=wrong-import-order, wrong-import-position
       from apache_beam.io import gcsio
       return gcsio.GcsIO().exists()
     else:
@@ -220,7 +220,7 @@ class ChannelFactory(object):
   @staticmethod
   def rmdir(path):
     if path.startswith('gs://'):
-      # pylint: disable=g-import-not-at-top
+      # pylint: disable=wrong-import-order, wrong-import-position
       from apache_beam.io import gcsio
       gcs = gcsio.GcsIO()
       if not path.endswith('/'):
@@ -237,7 +237,7 @@ class ChannelFactory(object):
   @staticmethod
   def rm(path):
     if path.startswith('gs://'):
-      # pylint: disable=g-import-not-at-top
+      # pylint: disable=wrong-import-order, wrong-import-position
       from apache_beam.io import gcsio
       gcsio.GcsIO().delete(path)
     else:
@@ -249,7 +249,7 @@ class ChannelFactory(object):
   @staticmethod
   def glob(path):
     if path.startswith('gs://'):
-      # pylint: disable=g-import-not-at-top
+      # pylint: disable=wrong-import-order, wrong-import-position
       from apache_beam.io import gcsio
       return gcsio.GcsIO().glob(path)
     else:
@@ -608,7 +608,7 @@ class TextFileReader(iobase.NativeSourceReader):
 
   def __enter__(self):
     if self.source.is_gcs_source:
-      # pylint: disable=g-import-not-at-top
+      # pylint: disable=wrong-import-order, wrong-import-position
       from apache_beam.io import gcsio
       self._file = gcsio.GcsIO().open(self.source.file_path, 'rb')
     else:

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/io/gcsio.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/io/gcsio.py b/sdks/python/apache_beam/io/gcsio.py
index 3d7e00f..a01988b 100644
--- a/sdks/python/apache_beam/io/gcsio.py
+++ b/sdks/python/apache_beam/io/gcsio.py
@@ -30,16 +30,17 @@ import re
 import StringIO
 import threading
 
+from apitools.base.py.exceptions import HttpError
+import apitools.base.py.transfer as transfer
+
 from apache_beam.internal import auth
 from apache_beam.utils import retry
 
-from apitools.base.py.exceptions import HttpError
-import apitools.base.py.transfer as transfer
 
 # Issue a friendlier error message if the storage library is not available.
 # TODO(silviuc): Remove this guard when storage is available everywhere.
 try:
-  # pylint: disable=g-import-not-at-top
+  # pylint: disable=wrong-import-order, wrong-import-position
   from apache_beam.internal.clients import storage
 except ImportError:
   raise RuntimeError(

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/io/gcsio_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/io/gcsio_test.py b/sdks/python/apache_beam/io/gcsio_test.py
index 138f2a8..eeabb1a 100644
--- a/sdks/python/apache_beam/io/gcsio_test.py
+++ b/sdks/python/apache_beam/io/gcsio_test.py
@@ -24,13 +24,12 @@ import random
 import threading
 import unittest
 
-
 import httplib2
-
-from apache_beam.io import gcsio
 from apitools.base.py.exceptions import HttpError
 from apache_beam.internal.clients import storage
 
+from apache_beam.io import gcsio
+
 
 class FakeGcsClient(object):
   # Fake storage client.  Usage in gcsio.py is client.objects.Get(...) and

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/pvalue.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/pvalue.py b/sdks/python/apache_beam/pvalue.py
index 6354139..78ff209 100644
--- a/sdks/python/apache_beam/pvalue.py
+++ b/sdks/python/apache_beam/pvalue.py
@@ -348,7 +348,7 @@ def AsSingleton(pcoll, default_value=_SINGLETON_NO_DEFAULT, label=None):  # pyli
   # Local import is required due to dependency loop; even though the
   # implementation of this function requires concepts defined in modules that
   # depend on pvalue, it lives in this module to reduce user workload.
-  from apache_beam.transforms import sideinputs  # pylint: disable=g-import-not-at-top
+  from apache_beam.transforms import sideinputs  # pylint: disable=wrong-import-order, wrong-import-position
   view = (pcoll | sideinputs.ViewAsSingleton(has_default, default_value,
                                              label=label))
   _cache_view(pcoll.pipeline, cache_key, view)
@@ -380,7 +380,7 @@ def AsIter(pcoll, label=None):  # pylint: disable=invalid-name
   # Local import is required due to dependency loop; even though the
   # implementation of this function requires concepts defined in modules that
   # depend on pvalue, it lives in this module to reduce user workload.
-  from apache_beam.transforms import sideinputs  # pylint: disable=g-import-not-at-top
+  from apache_beam.transforms import sideinputs  # pylint: disable=wrong-import-order, wrong-import-position
   view = (pcoll | sideinputs.ViewAsIterable(label=label))
   _cache_view(pcoll.pipeline, cache_key, view)
   return view
@@ -411,7 +411,7 @@ def AsList(pcoll, label=None):  # pylint: disable=invalid-name
   # Local import is required due to dependency loop; even though the
   # implementation of this function requires concepts defined in modules that
   # depend on pvalue, it lives in this module to reduce user workload.
-  from apache_beam.transforms import sideinputs  # pylint: disable=g-import-not-at-top
+  from apache_beam.transforms import sideinputs  # pylint: disable=wrong-import-order, wrong-import-position
   view = (pcoll | sideinputs.ViewAsList(label=label))
   _cache_view(pcoll.pipeline, cache_key, view)
   return view
@@ -443,7 +443,7 @@ def AsDict(pcoll, label=None):  # pylint: disable=invalid-name
   # Local import is required due to dependency loop; even though the
   # implementation of this function requires concepts defined in modules that
   # depend on pvalue, it lives in this module to reduce user workload.
-  from apache_beam.transforms import sideinputs  # pylint: disable=g-import-not-at-top
+  from apache_beam.transforms import sideinputs  # pylint: disable=wrong-import-order, wrong-import-position
   view = (pcoll | sideinputs.ViewAsDict(label=label))
   _cache_view(pcoll.pipeline, cache_key, view)
   return view

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/runners/common.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/common.py b/sdks/python/apache_beam/runners/common.py
index 0e4a057..e33d4ce 100644
--- a/sdks/python/apache_beam/runners/common.py
+++ b/sdks/python/apache_beam/runners/common.py
@@ -32,8 +32,10 @@ from apache_beam.transforms.window import WindowFn
 class FakeLogger(object):
   def PerThreadLoggingContext(self, *unused_args, **unused_kwargs):
     return self
+
   def __enter__(self):
     pass
+
   def __exit__(self, *unused_args):
     pass
 
@@ -153,18 +155,22 @@ class DoFnRunner(object):
       else:
         self.tagged_receivers[tag].output(windowed_value)
 
+
 class NoContext(WindowFn.AssignContext):
   """An uninspectable WindowFn.AssignContext."""
   NO_VALUE = object()
+
   def __init__(self, value, timestamp=NO_VALUE):
     self.value = value
     self._timestamp = timestamp
+
   @property
   def timestamp(self):
     if self._timestamp is self.NO_VALUE:
       raise ValueError('No timestamp in this context.')
     else:
       return self._timestamp
+
   @property
   def existing_windows(self):
     raise ValueError('No existing_windows in this context.')

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/runners/dataflow_runner.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/dataflow_runner.py b/sdks/python/apache_beam/runners/dataflow_runner.py
index 7b04ae9..43a50ba 100644
--- a/sdks/python/apache_beam/runners/dataflow_runner.py
+++ b/sdks/python/apache_beam/runners/dataflow_runner.py
@@ -86,6 +86,7 @@ class DataflowPipelineRunner(PipelineRunner):
     # It typically takes about 30 seconds.
     final_countdown_timer_secs = 50.0
     sleep_secs = 5.0
+
     # Try to prioritize the user-level traceback, if any.
     def rank_error(msg):
       if 'work item was attempted' in msg:
@@ -151,7 +152,7 @@ class DataflowPipelineRunner(PipelineRunner):
   def run(self, pipeline):
     """Remotely executes entire pipeline or parts reachable from node."""
     # Import here to avoid adding the dependency for local running scenarios.
-    # pylint: disable=g-import-not-at-top
+    # pylint: disable=wrong-import-order, wrong-import-position
     from apache_beam.internal import apiclient
     self.job = apiclient.Job(pipeline.options)
     # The superclass's run will trigger a traversal of all reachable nodes.
@@ -244,7 +245,7 @@ class DataflowPipelineRunner(PipelineRunner):
   def _add_step(self, step_kind, step_label, transform_node, side_tags=()):
     """Creates a Step object and adds it to the cache."""
     # Import here to avoid adding the dependency for local running scenarios.
-    # pylint: disable=g-import-not-at-top
+    # pylint: disable=wrong-import-order, wrong-import-position
     from apache_beam.internal import apiclient
     step = apiclient.Step(step_kind, self._get_unique_step_name())
     self.job.proto.steps.append(step.proto)

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/runners/direct_runner.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/direct_runner.py b/sdks/python/apache_beam/runners/direct_runner.py
index 2c73394..e0df439 100644
--- a/sdks/python/apache_beam/runners/direct_runner.py
+++ b/sdks/python/apache_beam/runners/direct_runner.py
@@ -112,7 +112,7 @@ class DirectPipelineRunner(PipelineRunner):
     values = self._cache.get_pvalue(transform_node.inputs[0])
     if isinstance(view, SingletonPCollectionView):
       has_default, default_value = view._view_options()  # pylint: disable=protected-access
-      if len(values) == 0:  # pylint: disable=g-explicit-length-test
+      if len(values) == 0:
         if has_default:
           result = default_value
         else:

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/runners/runner.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/runner.py b/sdks/python/apache_beam/runners/runner.py
index 9b8d622..55b63f3 100644
--- a/sdks/python/apache_beam/runners/runner.py
+++ b/sdks/python/apache_beam/runners/runner.py
@@ -40,7 +40,7 @@ def create_runner(runner_name):
   Raises:
     RuntimeError: if an invalid runner name is used.
   """
-  # pylint: disable=g-import-not-at-top
+  # pylint: disable=wrong-import-order, wrong-import-position
   if runner_name == 'DirectPipelineRunner':
     import apache_beam.runners.direct_runner
     return apache_beam.runners.direct_runner.DirectPipelineRunner()
@@ -81,7 +81,7 @@ class PipelineRunner(object):
     """Execute the entire pipeline or the sub-DAG reachable from a node."""
 
     # Imported here to avoid circular dependencies.
-    # pylint: disable=g-import-not-at-top
+    # pylint: disable=wrong-import-order, wrong-import-position
     from apache_beam.pipeline import PipelineVisitor
 
     class RunVisitor(PipelineVisitor):
@@ -117,7 +117,7 @@ class PipelineRunner(object):
     """
 
     # Imported here to avoid circular dependencies.
-    # pylint: disable=g-import-not-at-top
+    # pylint: disable=wrong-import-order, wrong-import-position
     from apache_beam.pipeline import PipelineVisitor
 
     class ClearVisitor(PipelineVisitor):

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/transforms/aggregator.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/aggregator.py b/sdks/python/apache_beam/transforms/aggregator.py
index b7167ff..a5e83cb 100644
--- a/sdks/python/apache_beam/transforms/aggregator.py
+++ b/sdks/python/apache_beam/transforms/aggregator.py
@@ -103,6 +103,6 @@ class Aggregator(object):
 
 
 def _is_supported_kind(combine_fn):
-  # pylint: disable=g-import-not-at-top
+  # pylint: disable=wrong-import-order, wrong-import-position
   from apache_beam.internal.apiclient import metric_translations
   return combine_fn.__class__ in metric_translations

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/transforms/combiners.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/combiners.py b/sdks/python/apache_beam/transforms/combiners.py
index 64ede3b..e9f11a0 100644
--- a/sdks/python/apache_beam/transforms/combiners.py
+++ b/sdks/python/apache_beam/transforms/combiners.py
@@ -46,6 +46,11 @@ __all__ = [
     'ToList',
     ]
 
+# Type variables
+T = TypeVariable('T')
+K = TypeVariable('K')
+V = TypeVariable('V')
+
 
 class Mean(object):
   """Combiners for computing arithmetic means of elements."""
@@ -214,7 +219,6 @@ class Top(object):
     return pcoll | Top.PerKey(label, n, lambda a, b: b < a)
 
 
-T = TypeVariable('T')
 @with_input_types(T)
 @with_output_types(List[T])
 class TopCombineFn(core.CombineFn):
@@ -329,7 +333,6 @@ class Sample(object):
     return pcoll | core.CombinePerKey(label, SampleCombineFn(n))
 
 
-T = TypeVariable('T')
 @with_input_types(T)
 @with_output_types(List[T])
 class SampleCombineFn(core.CombineFn):
@@ -404,7 +407,6 @@ class ToList(ptransform.PTransform):
     return pcoll | core.CombineGlobally(self.label, ToListCombineFn())
 
 
-T = TypeVariable('T')
 @with_input_types(T)
 @with_output_types(List[T])
 class ToListCombineFn(core.CombineFn):
@@ -439,8 +441,6 @@ class ToDict(ptransform.PTransform):
     return pcoll | core.CombineGlobally(self.label, ToDictCombineFn())
 
 
-K = TypeVariable('K')
-V = TypeVariable('V')
 @with_input_types(Tuple[K, V])
 @with_output_types(Dict[K, V])
 class ToDictCombineFn(core.CombineFn):

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/transforms/combiners_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/combiners_test.py b/sdks/python/apache_beam/transforms/combiners_test.py
index 112c591..10682b4 100644
--- a/sdks/python/apache_beam/transforms/combiners_test.py
+++ b/sdks/python/apache_beam/transforms/combiners_test.py
@@ -122,6 +122,7 @@ class CombineTest(unittest.TestCase):
       pipeline = Pipeline('DirectPipelineRunner')
       pcoll = pipeline | Create('start', [1, 1, 2, 2])
       result = pcoll | combine.Sample.FixedSizeGlobally('sample-%d' % ix, 3)
+
       def matcher():
         def match(actual):
           # There is always exactly one result.
@@ -142,6 +143,7 @@ class CombineTest(unittest.TestCase):
         'start-perkey',
         sum(([(i, 1), (i, 1), (i, 2), (i, 2)] for i in xrange(300)), []))
     result = pcoll | combine.Sample.FixedSizePerKey('sample', 3)
+
     def matcher():
       def match(actual):
         for _, samples in actual:
@@ -180,6 +182,7 @@ class CombineTest(unittest.TestCase):
     the_list = [6, 3, 1, 1, 9, 1, 5, 2, 0, 6]
     pcoll = pipeline | Create('start', the_list)
     result = pcoll | combine.ToList('to list')
+
     def matcher(expected):
       def match(actual):
         equal_to(expected[0])(actual[0])
@@ -191,6 +194,7 @@ class CombineTest(unittest.TestCase):
     pairs = [(1, 2), (3, 4), (5, 6)]
     pcoll = pipeline | Create('start-pairs', pairs)
     result = pcoll | combine.ToDict('to dict')
+
     def matcher():
       def match(actual):
         equal_to([1])([len(actual)])

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/transforms/core.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/core.py b/sdks/python/apache_beam/transforms/core.py
index 445367f..8f4c246 100644
--- a/sdks/python/apache_beam/transforms/core.py
+++ b/sdks/python/apache_beam/transforms/core.py
@@ -49,6 +49,11 @@ from apache_beam.typehints import WithTypeHints
 from apache_beam.typehints.trivial_inference import element_type
 from apache_beam.utils.options import TypeOptions
 
+# Type variables
+T = typehints.TypeVariable('T')
+K = typehints.TypeVariable('K')
+V = typehints.TypeVariable('V')
+
 
 class DoFnProcessContext(object):
   """A processing context passed to DoFn methods during execution.
@@ -842,6 +847,7 @@ class CombineGlobally(PTransform):
             "an empty PCollection if the input PCollection is empty, "
             "or CombineGlobally().as_singleton_view() to get the default "
             "output of the CombineFn if the input PCollection is empty.")
+
       def typed(transform):
         # TODO(robertwb): We should infer this.
         if combined.element_type:
@@ -954,8 +960,6 @@ class CombineValuesDoFn(DoFn):
     return hints
 
 
-K = typehints.TypeVariable('K')
-V = typehints.TypeVariable('V')
 @typehints.with_input_types(typehints.KV[K, V])
 @typehints.with_output_types(typehints.KV[K, typehints.Iterable[V]])
 class GroupByKey(PTransform):
@@ -999,10 +1003,10 @@ class GroupByKey(PTransform):
 
     def process(self, context):
       k, vs = context.element
-      # pylint: disable=g-import-not-at-top
+      # pylint: disable=wrong-import-order, wrong-import-position
       from apache_beam.transforms.trigger import InMemoryUnmergedState
       from apache_beam.transforms.trigger import create_trigger_driver
-      # pylint: enable=g-import-not-at-top
+      # pylint: enable=wrong-import-order, wrong-import-position
       driver = create_trigger_driver(self.windowing, True)
       state = InMemoryUnmergedState()
       # TODO(robertwb): Conditionally process in smaller chunks.
@@ -1051,8 +1055,6 @@ class GroupByKey(PTransform):
                       self.GroupAlsoByWindow(pcoll.windowing)))
 
 
-K = typehints.TypeVariable('K')
-V = typehints.TypeVariable('V')
 @typehints.with_input_types(typehints.KV[K, V])
 @typehints.with_output_types(typehints.KV[K, typehints.Iterable[V]])
 class GroupByKeyOnly(PTransform):
@@ -1115,9 +1117,9 @@ class Windowing(object):
   def __init__(self, windowfn, triggerfn=None, accumulation_mode=None,
                output_time_fn=None):
     global AccumulationMode, DefaultTrigger
-    # pylint: disable=g-import-not-at-top
+    # pylint: disable=wrong-import-order, wrong-import-position
     from apache_beam.transforms.trigger import AccumulationMode, DefaultTrigger
-    # pylint: enable=g-import-not-at-top
+    # pylint: enable=wrong-import-order, wrong-import-position
     if triggerfn is None:
       triggerfn = DefaultTrigger()
     if accumulation_mode is None:
@@ -1145,10 +1147,9 @@ class Windowing(object):
     return self._is_default
 
 
-T = typehints.TypeVariable('T')
 @typehints.with_input_types(T)
 @typehints.with_output_types(T)
-class WindowInto(ParDo):  # pylint: disable=g-wrong-blank-lines
+class WindowInto(ParDo):
   """A window transform assigning windows to each element of a PCollection.
 
   Transforms an input PCollection by applying a windowing function to each

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/transforms/cy_combiners.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/cy_combiners.py b/sdks/python/apache_beam/transforms/cy_combiners.py
index be40d66..f824870 100644
--- a/sdks/python/apache_beam/transforms/cy_combiners.py
+++ b/sdks/python/apache_beam/transforms/cy_combiners.py
@@ -26,20 +26,25 @@ class AccumulatorCombineFn(core.CombineFn):
   # singleton?
   def create_accumulator(self):
     return self._accumulator_type()
+
   @staticmethod
   def add_input(accumulator, element):
     accumulator.add_input(element)
     return accumulator
+
   def merge_accumulators(self, accumulators):
     accumulator = self._accumulator_type()
     accumulator.merge(accumulators)
     return accumulator
+
   @staticmethod
   def extract_output(accumulator):
     return accumulator.extract_output()
+
   def __eq__(self, other):
     return (isinstance(other, AccumulatorCombineFn)
             and self._accumulator_type is other._accumulator_type)
+
   def __hash__(self):
     return hash(self._accumulator_type)
 
@@ -52,11 +57,14 @@ globals()['INT64_MIN'] = -2**_63
 class CountAccumulator(object):
   def __init__(self):
     self.value = 0
+
   def add_input(self, unused_element):
     self.value += 1
+
   def merge(self, accumulators):
     for accumulator in accumulators:
       self.value += accumulator.value
+
   def extract_output(self):
     return self.value
 
@@ -64,14 +72,17 @@ class CountAccumulator(object):
 class SumInt64Accumulator(object):
   def __init__(self):
     self.value = 0
+
   def add_input(self, element):
     element = int(element)
     if not INT64_MIN <= element <= INT64_MAX:
       raise OverflowError(element)
     self.value += element
+
   def merge(self, accumulators):
     for accumulator in accumulators:
       self.value += accumulator.value
+
   def extract_output(self):
     if not INT64_MIN <= self.value <= INT64_MAX:
       self.value %= 2**64
@@ -83,16 +94,19 @@ class SumInt64Accumulator(object):
 class MinInt64Accumulator(object):
   def __init__(self):
     self.value = INT64_MAX
+
   def add_input(self, element):
     element = int(element)
     if not INT64_MIN <= element <= INT64_MAX:
       raise OverflowError(element)
     if element < self.value:
       self.value = element
+
   def merge(self, accumulators):
     for accumulator in accumulators:
       if accumulator.value < self.value:
         self.value = accumulator.value
+
   def extract_output(self):
     return self.value
 
@@ -100,16 +114,19 @@ class MinInt64Accumulator(object):
 class MaxInt64Accumulator(object):
   def __init__(self):
     self.value = INT64_MIN
+
   def add_input(self, element):
     element = int(element)
     if not INT64_MIN <= element <= INT64_MAX:
       raise OverflowError(element)
     if element > self.value:
       self.value = element
+
   def merge(self, accumulators):
     for accumulator in accumulators:
       if accumulator.value > self.value:
         self.value = accumulator.value
+
   def extract_output(self):
     return self.value
 
@@ -118,16 +135,19 @@ class MeanInt64Accumulator(object):
   def __init__(self):
     self.sum = 0
     self.count = 0
+
   def add_input(self, element):
     element = int(element)
     if not INT64_MIN <= element <= INT64_MAX:
       raise OverflowError(element)
     self.sum += element
     self.count += 1
+
   def merge(self, accumulators):
     for accumulator in accumulators:
       self.sum += accumulator.sum
       self.count += accumulator.count
+
   def extract_output(self):
     if not INT64_MIN <= self.sum <= INT64_MAX:
       self.sum %= 2**64
@@ -138,12 +158,20 @@ class MeanInt64Accumulator(object):
 
 class CountCombineFn(AccumulatorCombineFn):
   _accumulator_type = CountAccumulator
+
+
 class SumInt64Fn(AccumulatorCombineFn):
   _accumulator_type = SumInt64Accumulator
+
+
 class MinInt64Fn(AccumulatorCombineFn):
   _accumulator_type = MinInt64Accumulator
+
+
 class MaxInt64Fn(AccumulatorCombineFn):
   _accumulator_type = MaxInt64Accumulator
+
+
 class MeanInt64Fn(AccumulatorCombineFn):
   _accumulator_type = MeanInt64Accumulator
 
@@ -156,12 +184,15 @@ _NAN = float('nan')
 class SumDoubleAccumulator(object):
   def __init__(self):
     self.value = 0
+
   def add_input(self, element):
     element = float(element)
     self.value += element
+
   def merge(self, accumulators):
     for accumulator in accumulators:
       self.value += accumulator.value
+
   def extract_output(self):
     return self.value
 
@@ -169,14 +200,17 @@ class SumDoubleAccumulator(object):
 class MinDoubleAccumulator(object):
   def __init__(self):
     self.value = _POS_INF
+
   def add_input(self, element):
     element = float(element)
     if element < self.value:
       self.value = element
+
   def merge(self, accumulators):
     for accumulator in accumulators:
       if accumulator.value < self.value:
         self.value = accumulator.value
+
   def extract_output(self):
     return self.value
 
@@ -184,14 +218,17 @@ class MinDoubleAccumulator(object):
 class MaxDoubleAccumulator(object):
   def __init__(self):
     self.value = _NEG_INF
+
   def add_input(self, element):
     element = float(element)
     if element > self.value:
       self.value = element
+
   def merge(self, accumulators):
     for accumulator in accumulators:
       if accumulator.value > self.value:
         self.value = accumulator.value
+
   def extract_output(self):
     return self.value
 
@@ -200,24 +237,33 @@ class MeanDoubleAccumulator(object):
   def __init__(self):
     self.sum = 0
     self.count = 0
+
   def add_input(self, element):
     element = float(element)
     self.sum += element
     self.count += 1
+
   def merge(self, accumulators):
     for accumulator in accumulators:
       self.sum += accumulator.sum
       self.count += accumulator.count
+
   def extract_output(self):
     return self.sum / self.count if self.count else _NAN
 
 
 class SumFloatFn(AccumulatorCombineFn):
   _accumulator_type = SumDoubleAccumulator
+
+
 class MinFloatFn(AccumulatorCombineFn):
   _accumulator_type = MinDoubleAccumulator
+
+
 class MaxFloatFn(AccumulatorCombineFn):
   _accumulator_type = MaxDoubleAccumulator
+
+
 class MeanFloatFn(AccumulatorCombineFn):
   _accumulator_type = MeanDoubleAccumulator
 
@@ -225,11 +271,14 @@ class MeanFloatFn(AccumulatorCombineFn):
 class AllAccumulator(object):
   def __init__(self):
     self.value = True
+
   def add_input(self, element):
     self.value &= not not element
+
   def merge(self, accumulators):
     for accumulator in accumulators:
       self.value &= accumulator.value
+
   def extract_output(self):
     return self.value
 
@@ -237,11 +286,14 @@ class AllAccumulator(object):
 class AnyAccumulator(object):
   def __init__(self):
     self.value = False
+
   def add_input(self, element):
     self.value |= not not element
+
   def merge(self, accumulators):
     for accumulator in accumulators:
       self.value |= accumulator.value
+
   def extract_output(self):
     return self.value
 
@@ -249,5 +301,6 @@ class AnyAccumulator(object):
 class AnyCombineFn(AccumulatorCombineFn):
   _accumulator_type = AnyAccumulator
 
+
 class AllCombineFn(AccumulatorCombineFn):
   _accumulator_type = AllAccumulator

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/transforms/ptransform.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/ptransform.py b/sdks/python/apache_beam/transforms/ptransform.py
index 5b089b9..106c44f 100644
--- a/sdks/python/apache_beam/transforms/ptransform.py
+++ b/sdks/python/apache_beam/transforms/ptransform.py
@@ -383,10 +383,10 @@ class PTransform(WithTypeHints):
     pipelines = [v.pipeline for v in pvalues if isinstance(v, pvalue.PValue)]
     if pvalues and not pipelines:
       deferred = False
-      # pylint: disable=g-import-not-at-top
+      # pylint: disable=wrong-import-order, wrong-import-position
       from apache_beam import pipeline
       from apache_beam.utils.options import PipelineOptions
-      # pylint: enable=g-import-not-at-top
+      # pylint: enable=wrong-import-order, wrong-import-position
       p = pipeline.Pipeline(
           'DirectPipelineRunner', PipelineOptions(sys.argv))
     else:
@@ -403,9 +403,9 @@ class PTransform(WithTypeHints):
             raise ValueError(
                 'Mixing value from different pipelines not allowed.')
       deferred = not getattr(p.runner, 'is_eager', False)
-    # pylint: disable=g-import-not-at-top
+    # pylint: disable=wrong-import-order, wrong-import-position
     from apache_beam.transforms.core import Create
-    # pylint: enable=g-import-not-at-top
+    # pylint: enable=wrong-import-order, wrong-import-position
     replacements = {id(v): p | Create('CreatePInput%s' % ix, v)
                     for ix, v in enumerate(pvalues)
                     if not isinstance(v, pvalue.PValue) and v is not None}
@@ -431,9 +431,9 @@ class PTransform(WithTypeHints):
 
     Generally only needs to be overriden for multi-input PTransforms.
     """
-    # pylint: disable=g-import-not-at-top
+    # pylint: disable=wrong-import-order
     from apache_beam import pipeline
-    # pylint: enable=g-import-not-at-top
+    # pylint: enable=wrong-import-order
     if isinstance(pvalueish, pipeline.Pipeline):
       pvalueish = pvalue.PBegin(pvalueish)
 
@@ -557,6 +557,7 @@ class PTransformWithSideInputs(PTransform):
     type_hints = self.get_type_hints().input_types
     if type_hints:
       args, kwargs = self.raw_side_inputs
+
       def element_type(side_input):
         if isinstance(side_input, pvalue.PCollectionView):
           return side_input.element_type

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/transforms/ptransform_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/ptransform_test.py b/sdks/python/apache_beam/transforms/ptransform_test.py
index 8ae7a37..d6ee18a 100644
--- a/sdks/python/apache_beam/transforms/ptransform_test.py
+++ b/sdks/python/apache_beam/transforms/ptransform_test.py
@@ -208,8 +208,10 @@ class PTransformTest(unittest.TestCase):
     class MyDoFn(beam.DoFn):
       def start_bundle(self, c):
         yield 'start'
+
       def process(self, c):
         pass
+
       def finish_bundle(self, c):
         yield 'finish'
     pipeline = Pipeline('DirectPipelineRunner')
@@ -540,6 +542,7 @@ class PTransformTest(unittest.TestCase):
       def _extract_input_pvalues(self, pvalueish):
         pvalueish = list(pvalueish)
         return pvalueish, sum([list(p.values()) for p in pvalueish], [])
+
       def apply(self, pcoll_dicts):
         keys = reduce(operator.or_, [set(p.keys()) for p in pcoll_dicts])
         res = {}
@@ -554,6 +557,7 @@ class PTransformTest(unittest.TestCase):
     self.assertEqual(['x', 'x', 'y', 'y', 'z'], sorted(res['b']))
     self.assertEqual([], sorted(res['c']))
 
+
 @beam.ptransform_fn
 def SamplePTransform(label, pcoll, context, *args, **kwargs):
   """Sample transform using the @ptransform_fn decorator."""
@@ -1646,6 +1650,7 @@ class PTransformTypeCheckTestCase(TypeHintTestCase):
          | combine.Sample.FixedSizeGlobally('sample', 3))
 
     self.assertCompatible(typehints.Iterable[int], d.element_type)
+
     def matcher(expected_len):
       def match(actual):
         equal_to([expected_len])([len(actual[0])])
@@ -1661,6 +1666,7 @@ class PTransformTypeCheckTestCase(TypeHintTestCase):
          | combine.Sample.FixedSizeGlobally('sample', 2))
 
     self.assertCompatible(typehints.Iterable[int], d.element_type)
+
     def matcher(expected_len):
       def match(actual):
         equal_to([expected_len])([len(actual[0])])
@@ -1676,6 +1682,7 @@ class PTransformTypeCheckTestCase(TypeHintTestCase):
 
     self.assertCompatible(typehints.KV[int, typehints.Iterable[int]],
                           d.element_type)
+
     def matcher(expected_len):
       def match(actual):
         for _, sample in actual:
@@ -1694,6 +1701,7 @@ class PTransformTypeCheckTestCase(TypeHintTestCase):
 
     self.assertCompatible(typehints.KV[int, typehints.Iterable[int]],
                           d.element_type)
+
     def matcher(expected_len):
       def match(actual):
         for _, sample in actual:
@@ -1708,6 +1716,7 @@ class PTransformTypeCheckTestCase(TypeHintTestCase):
          | combine.ToList('to list'))
 
     self.assertCompatible(typehints.List[int], d.element_type)
+
     def matcher(expected):
       def match(actual):
         equal_to(expected)(actual[0])
@@ -1723,6 +1732,7 @@ class PTransformTypeCheckTestCase(TypeHintTestCase):
          | combine.ToList('to list'))
 
     self.assertCompatible(typehints.List[str], d.element_type)
+
     def matcher(expected):
       def match(actual):
         equal_to(expected)(actual[0])

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/transforms/sideinputs.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/sideinputs.py b/sdks/python/apache_beam/transforms/sideinputs.py
index fc54fa3..6484a7c 100644
--- a/sdks/python/apache_beam/transforms/sideinputs.py
+++ b/sdks/python/apache_beam/transforms/sideinputs.py
@@ -28,6 +28,10 @@ from apache_beam import pvalue
 from apache_beam import typehints
 from apache_beam.transforms.ptransform import PTransform
 
+# Type variables
+K = typehints.TypeVariable('K')
+V = typehints.TypeVariable('V')
+
 
 class CreatePCollectionView(PTransform):
   """Transform to materialize a given PCollectionView in the pipeline.
@@ -119,11 +123,10 @@ class ViewAsList(PTransform):
             .with_input_types(input_type)
             .with_output_types(output_type))
 
-K = typehints.TypeVariable('K')
-V = typehints.TypeVariable('V')
+
 @typehints.with_input_types(typehints.Tuple[K, V])
 @typehints.with_output_types(typehints.Dict[K, V])
-class ViewAsDict(PTransform):  # pylint: disable=g-wrong-blank-lines
+class ViewAsDict(PTransform):
   """Transform to view PCollection as a dict PCollectionView.
 
   Important: this transform is an implementation detail and should not be used

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/transforms/trigger.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/trigger.py b/sdks/python/apache_beam/transforms/trigger.py
index 1c75d3b..788277c 100644
--- a/sdks/python/apache_beam/transforms/trigger.py
+++ b/sdks/python/apache_beam/transforms/trigger.py
@@ -731,6 +731,7 @@ class DefaultGlobalBatchTriggerDriver(TriggerDriver):
             unwindowed_value = wv.value
             self.notify_observers(unwindowed_value)
             yield unwindowed_value
+
         def __repr__(self):
           return '<UnwindowedValues of %s>' % windowed_values
       unwindowed = UnwindowedValues()

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/transforms/trigger_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/trigger_test.py b/sdks/python/apache_beam/transforms/trigger_test.py
index fee9b1b..2cdd1e3 100644
--- a/sdks/python/apache_beam/transforms/trigger_test.py
+++ b/sdks/python/apache_beam/transforms/trigger_test.py
@@ -61,6 +61,7 @@ class TriggerTest(unittest.TestCase):
                          **kwargs):
     late_data = kwargs.pop('late_data', [])
     assert not kwargs
+
     def bundle_data(data, size):
       bundle = []
       for timestamp, elem in data:
@@ -477,10 +478,10 @@ class TranscriptTest(unittest.TestCase):
       else:
         return fn
 
-    # pylint: disable=g-import-not-at-top
+    # pylint: disable=wrong-import-order, wrong-import-position
     from apache_beam.transforms import window as window_module
     from apache_beam.transforms import trigger as trigger_module
-    # pylint: enable=g-import-not-at-top
+    # pylint: enable=wrong-import-order, wrong-import-position
     window_fn_names = dict(window_module.__dict__)
     window_fn_names.update({'CustomTimestampingFixedWindowsWindowFn':
                             CustomTimestampingFixedWindowsWindowFn})

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/transforms/util.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/util.py b/sdks/python/apache_beam/transforms/util.py
index 52bba05..b8380a0 100644
--- a/sdks/python/apache_beam/transforms/util.py
+++ b/sdks/python/apache_beam/transforms/util.py
@@ -181,6 +181,7 @@ class DataflowAssertException(Exception):
 # TODO(silviuc): Add contains_in_any_order-style matchers.
 def equal_to(expected):
   expected = list(expected)
+
   def _equal(actual):
     sorted_expected = sorted(expected)
     sorted_actual = sorted(actual)

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/transforms/window_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/window_test.py b/sdks/python/apache_beam/transforms/window_test.py
index 1d7f272..186fcd4 100644
--- a/sdks/python/apache_beam/transforms/window_test.py
+++ b/sdks/python/apache_beam/transforms/window_test.py
@@ -52,6 +52,7 @@ class ReifyWindowsFn(core.DoFn):
       yield "%s @ %s" % (key, window), values
 reify_windows = core.ParDo(ReifyWindowsFn())
 
+
 class WindowTest(unittest.TestCase):
 
   def test_fixed_windows(self):

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/typehints/typehints_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/typehints/typehints_test.py b/sdks/python/apache_beam/typehints/typehints_test.py
index e6bab61..aa04fe2 100644
--- a/sdks/python/apache_beam/typehints/typehints_test.py
+++ b/sdks/python/apache_beam/typehints/typehints_test.py
@@ -45,6 +45,7 @@ def check_or_interleave(hint, value, var):
     _check_instance_type(hint, value, var)
     return value
 
+
 def check_type_hints(f):
   @functools.wraps(f)
   def wrapper(*args, **kwargs):
@@ -770,6 +771,7 @@ class TakesDecoratorTestCase(TypeHintTestCase):
   def test_must_be_primitive_type_or_constraint(self):
     with self.assertRaises(TypeError) as e:
       t = [1, 2]
+
       @with_input_types(a=t)
       def foo(a):
         pass
@@ -781,6 +783,7 @@ class TakesDecoratorTestCase(TypeHintTestCase):
 
     with self.assertRaises(TypeError) as e:
       t = 5
+
       @check_type_hints
       @with_input_types(a=t)
       def foo(a):

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/utils/dependency_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/utils/dependency_test.py b/sdks/python/apache_beam/utils/dependency_test.py
index 4edcaa7..8a97f4b 100644
--- a/sdks/python/apache_beam/utils/dependency_test.py
+++ b/sdks/python/apache_beam/utils/dependency_test.py
@@ -336,6 +336,7 @@ class SetupTest(unittest.TestCase):
         'gs://my-gcs-bucket/gcs.tar.gz']
 
     gcs_copied_files = []
+
     def file_copy(from_path, to_path):
       if from_path.startswith('gs://'):
         gcs_copied_files.append(from_path)

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/apache_beam/utils/retry_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/utils/retry_test.py b/sdks/python/apache_beam/utils/retry_test.py
index 0a016b9..705c555 100644
--- a/sdks/python/apache_beam/utils/retry_test.py
+++ b/sdks/python/apache_beam/utils/retry_test.py
@@ -19,10 +19,10 @@
 
 import unittest
 
-from apache_beam.utils import retry
-
 from apitools.base.py.exceptions import HttpError
 
+from apache_beam.utils import retry
+
 
 class FakeClock(object):
   """A fake clock object implementing sleep() and recording calls."""

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/run_pylint.sh
----------------------------------------------------------------------
diff --git a/sdks/python/run_pylint.sh b/sdks/python/run_pylint.sh
index e8062c0..4e0b129 100755
--- a/sdks/python/run_pylint.sh
+++ b/sdks/python/run_pylint.sh
@@ -16,10 +16,11 @@
 #    limitations under the License.
 #
 
-# This script will run pylint on files that changed compared to the current
-# HEAD of the branch.
+# This script will run pylint and pep8 on files that changed compared to the
+# current HEAD of the branch.
 #
 # Use "pylint apache_beam" to run pylint all files.
+# Use "pep8 apache_beam" to run pep8 all files.
 #
 # The exit-code of the script indicates success or a failure.
 
@@ -43,6 +44,8 @@ if test "$CHANGED_FILES"; then
   echo "Running pylint on changed files:"
   echo "$CHANGED_FILES"
   pylint $CHANGED_FILES
+  echo "Running pep8 on changed files:"
+  pep8 $CHANGED_FILES
 else
   echo "Not running pylint. No eligible files."
 fi

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/57a4495d/sdks/python/tox.ini
----------------------------------------------------------------------
diff --git a/sdks/python/tox.ini b/sdks/python/tox.ini
index 356de57..29674ed 100644
--- a/sdks/python/tox.ini
+++ b/sdks/python/tox.ini
@@ -18,8 +18,18 @@
 [tox]
 envlist = py27
 
+[pep8]
+# Disable all errors and warnings except for the ones related to blank lines.
+# pylint does not check the number of blank lines.
+select = E3
+
+# Skip auto generated files (windmill_pb2.py, windmill_service_pb2.py)
+exclude = windmill_pb2.py, windmill_service_pb2.py
+
 [testenv:py27]
-deps=pylint
+deps=
+  pep8
+  pylint
 commands =
   python setup.py test
   {toxinidir}/run_pylint.sh


Mime
View raw message