fixed E501 (#32)

Author: liferoad

Diff for: Python/.flake8

@@ -5,7 +5,6 @@ ignore =
   E203
   W503
   F841
-  E501
 exclude =
     .eggs
     .git

Diff for: Python/.isort.cfg

@@ -18,7 +18,6 @@ import_heading_stdlib=standard libraries
 import_heading_thirdparty=third party libraries
 include_trailing_comma=True
 indent='    '
-known_dfml=src
 dedup_headings=True
 line_length=80
 multi_line_output=3

Diff for: Python/basics/combine_globally.py

@@ -24,7 +24,7 @@ def run(argv=None):
     elements = [
         "Lorem ipsum dolor sit amet. Consectetur adipiscing elit",
         "Sed eu velit nec sem vulputate loborti",
-        "In lobortis augue vitae sagittis molestie. Mauris volutpat tortor non purus elementum",
+        "In lobortis augue vitae sagittis molestie. Mauris volutpat tortor non purus elementum",  # noqa:E501
         "Ut blandit massa et risus sollicitudin auctor",
     ]
 

Diff for: Python/basics/combine_per_key.py

@@ -24,22 +24,22 @@ def run(argv=None):
     elements = [
         (
             "Latin",
-            "Lorem ipsum dolor sit amet. Consectetur adipiscing elit. Sed eu velit nec sem vulputate loborti",
+            "Lorem ipsum dolor sit amet. Consectetur adipiscing elit. Sed eu velit nec sem vulputate loborti",  # noqa:E501
         ),
         (
             "Latin",
-            "In lobortis augue vitae sagittis molestie. Mauris volutpat tortor non purus elementum",
+            "In lobortis augue vitae sagittis molestie. Mauris volutpat tortor non purus elementum",  # noqa:E501
         ),
         ("English", "From fairest creatures we desire increase"),
         ("English", "That thereby beauty's rose might never die"),
         ("English", "But as the riper should by time decease"),
         (
             "Spanish",
-            "En un lugar de la Mancha, de cuyo nombre no quiero acordarme, no ha mucho",
+            "En un lugar de la Mancha, de cuyo nombre no quiero acordarme, no ha mucho",  # noqa:E501
         ),
         (
             "Spanish",
-            "tiempo que vivía un hidalgo de los de lanza en astillero, adarga antigua",
+            "tiempo que vivía un hidalgo de los de lanza en astillero, adarga antigua",  # noqa:E501
         ),
     ]
 

Diff for: Python/basics/pardo.py

@@ -30,7 +30,7 @@ def run(argv=None):
     elements = [
         "Lorem ipsum dolor sit amet. Consectetur adipiscing elit",
         "Sed eu velit nec sem vulputate loborti",
-        "In lobortis augue vitae sagittis molestie. Mauris volutpat tortor non ",
+        "In lobortis augue vitae sagittis molestie. Mauris volutpat tortor non ",  # noqa:E501
         "Ut blandit massa et risus sollicitudin auctor",
     ]
 

Diff for: Python/bigquery/failed_rows_bigquery.py

@@ -30,7 +30,8 @@
 
 
 def run(argv=None):
-    """This pipeline shows how to access the rows that failed being inserted to BigQuery"""
+    """This pipeline shows how to access the rows that failed being
+    inserted to BigQuery"""
     topic = "projects/pubsub-public-data/topics/taxirides-realtime"
 
     class FailedRowsOptions(PipelineOptions):

Diff for: Python/bigquery/nested_bigquery.py

@@ -44,7 +44,8 @@ def _add_argparse_args(cls, parser):
 
 
 def run(argv=None):
-    """This pipeline shows how to read, write and modify nested fields from BigQuery"""
+    """This pipeline shows how to read, write and modify nested fields
+    from BigQuery"""
 
     schema = {
         "fields": [

Diff for: Python/bigquery/read_query_bigquery.py

@@ -28,8 +28,8 @@ def _add_argparse_args(cls, parser):
         parser.add_argument(
             "--query",
             default=(
-                "SELECT repository_language, COUNT(repository_language) totalRepos"
-                " FROM `bigquery-public-data.samples.github_timeline` GROUP BY 1"
+                "SELECT repository_language, COUNT(repository_language) totalRepos"  # noqa:E501
+                " FROM `bigquery-public-data.samples.github_timeline` GROUP BY 1"  # noqa:E501
             ),
             help="BigQuery query to read data",
         )

Diff for: Python/bigquery/repeated_bigquery.py

@@ -43,7 +43,8 @@ def _add_argparse_args(cls, parser):
 
 
 def run(argv=None):
-    """This pipeline shows how to read, write and modify nested fields from BigQuery"""
+    """This pipeline shows how to read, write and modify nested fields
+    from BigQuery"""
 
     schema = {
         "fields": [

Diff for: Python/bigquery/streaming_inserts_bigquery.py

@@ -41,7 +41,7 @@ class StreamingInsertsOptions(PipelineOptions):
         def _add_argparse_args(cls, parser):
             parser.add_argument("--output_table", help="BQ Table to write")
 
-    table_schema = "ride_status:STRING, passenger_count:INTEGER, meter_reading:FLOAT, timestamp:STRING"
+    table_schema = "ride_status:STRING, passenger_count:INTEGER, meter_reading:FLOAT, timestamp:STRING"  # noqa:E501
     options = StreamingInsertsOptions()
     with beam.Pipeline(options=options) as p:
         output = (

Diff for: Python/bigquery/streaming_load_jobs_bigquery.py

@@ -47,7 +47,7 @@ def _add_argparse_args(cls, parser):
                 help="Frequency to trigger the load job",
             )
 
-    table_schema = "ride_status:STRING, passenger_count:INTEGER, meter_reading:FLOAT, timestamp:STRING"
+    table_schema = "ride_status:STRING, passenger_count:INTEGER, meter_reading:FLOAT, timestamp:STRING"  # noqa:E501
     options = StreamingLoadOptions()
     with beam.Pipeline(options=options) as p:
         output = (

Diff for: Python/extra_examples/template.py

@@ -28,7 +28,8 @@
 
 """
 
-This example shows how to use Classic Templates. Note that Flex Templates are the preferred method.
+This example shows how to use Classic Templates. Note that Flex Templates
+are the preferred method.
 """
 
 

Diff for: Python/gcs/read_all_textio_streaming.py

@@ -33,7 +33,7 @@ def _add_argparse_args(cls, parser):
                 "--topic",
                 description="Provide a topic to read from",
                 dest="topic",
-                help="You need to create this topic and the messages will be GCS paths.",
+                help="You need to create this topic and the messages will be GCS paths.",  # noqa:E501
             )
 
     options = ReadAllStreamingTextOptions(streaming=True)

Diff for: Python/gcs/read_avro.py

@@ -40,7 +40,7 @@ def _add_argparse_args(cls, parser):
             | "ReadAvro" >> ReadFromAvro(options.path)
             | "CheckRow"
             >> Map(
-                lambda row: f"The abbreviation of {row['name']} is {row['post_abbr']}"
+                lambda row: f"The abbreviation of {row['name']} is {row['post_abbr']}"  # noqa:E501
             )
             | Map(logging.info)
         )

Diff for: Python/gcs/read_parquet.py

@@ -40,7 +40,7 @@ def _add_argparse_args(cls, parser):
             | "ReadParquet" >> ReadFromParquet(options.path)
             | "CheckRow"
             >> Map(
-                lambda row: f"The abbreviation of {row['name']} is {row['post_abbr']}"
+                lambda row: f"The abbreviation of {row['name']} is {row['post_abbr']}"  # noqa:E501
             )
             | Map(logging.info)
         )

Diff for: Python/pubsub/read_pubsub_topic.py

@@ -28,7 +28,8 @@ class ReadPubSubOptions(PipelineOptions):
         def _add_argparse_args(cls, parser):
             parser.add_argument(
                 "--topic",
-                # Run on Dataflow or authenticate to not get 403 PermissionDenied
+                # Run on Dataflow or authenticate to not get
+                # 403 PermissionDenied
                 default="projects/pubsub-public-data/topics/taxirides-realtime",
                 help="PubSub topic to read",
             )

Diff for: Python/testing_windows/accumulating_fired_panes.py

@@ -85,7 +85,7 @@ def get_input_stream():
 def print_with_info(element, window=DoFn.WindowParam, pane=DoFn.PaneInfoParam):
     """Helper function to print window and pane info"""
     print(
-        f"Element: {element[1]} \n\t Window: {window} \n\t Pane Timing: {PaneInfoTiming.to_string(pane.timing)}"
+        f"Element: {element[1]} \n\t Window: {window} \n\t Pane Timing: {PaneInfoTiming.to_string(pane.timing)}"  # noqa:E501
     )
 
 
@@ -106,10 +106,12 @@ def run():
         | "Window into FixedWindows"
         >> WindowInto(
             # We are using the default triggering which triggers when watermark
-            # crosses the window end and whenever valid late element(s) are added
+            # crosses the window end and whenever valid late element(s)
+            # are added
             FixedWindows(size=window_size_seconds),
             allowed_lateness=window_allowed_lateness_seconds,
-            # Late elements are combined with previous elements within the window
+            # Late elements are combined with previous elements within
+            # the window
             accumulation_mode=AccumulationMode.ACCUMULATING,
         )
         | "Combine elements within pane by key" >> GroupByKey()
@@ -126,12 +128,18 @@ def run():
 """
 EXPLANATION
 1 - Elements "On-Time-X" arrive to the pipeline before window closes
-2 - Elements "Late-(1 to 4)" arrive after window closing, but within the allowed late time
-3 - Elements "Late-(5, 6)" also arrive after window closing, but within the allowed late time
-4 - Element "Outside-allowed-lateness" is discarded as it's outside the allowed lateness
-
-
-The first trigger contains elements (1), since they arrive before window closing
-The second trigger contains elements (2) and elements (1), since we are accumulating
-The third trigger contains only elements (3) but also (2) and (1) because of the accumulation
+2 - Elements "Late-(1 to 4)" arrive after window closing, but within
+    the allowed late time
+3 - Elements "Late-(5, 6)" also arrive after window closing, but within
+    the allowed late time
+4 - Element "Outside-allowed-lateness" is discarded as it's outside
+    the allowed lateness
+
+
+The first trigger contains elements (1), since they arrive before
+window closing
+The second trigger contains elements (2) and elements (1),
+since we are accumulating
+The third trigger contains only elements (3) but also (2) and (1)
+because of the accumulation
 """

Diff for: Python/testing_windows/discarding_fired_panes.py

@@ -99,8 +99,8 @@ def run():
         >> WindowInto(
             FixedWindows(size=window_size_seconds),
             allowed_lateness=window_allowed_lateness_seconds,
-            # Late elements are contained within their own pane and no correlation
-            # exists between the panes
+            # Late elements are contained within their own pane
+            # and no correlation exists between the panes
             accumulation_mode=AccumulationMode.DISCARDING,
         )
         | "Combine elements within pane by key" >> GroupByKey()
@@ -118,13 +118,20 @@ def run():
 
 EXPLANATION
   1 - Elements "On-Time-X" arrive to the pipeline before window closes
-  2 - Elements "Late-(1 to 4)" arrive after window closing, but within the allowed late time
-  3 - Elements "Late-(5, 6)" also arrive after window closing, but within the allowed late time
-  4 - Element "Outside-allowed-lateness" is discarded as it's outside the allowed lateness
-
-  The first trigger contains elements (1), since they arrive before window closing
-  The second trigger contains elements (2) and only elements (2). They are triggered by themselves as they arrived late
-  The thrid trigger contains only elements (3) since they also arrived late but at different time
-
-Element "On-Time-New-Window-1" shows that this process continues until termination
+  2 - Elements "Late-(1 to 4)" arrive after window closing,
+      but within the allowed late time
+  3 - Elements "Late-(5, 6)" also arrive after window closing,
+      but within the allowed late time
+  4 - Element "Outside-allowed-lateness" is discarded
+      as it's outside the allowed lateness
+
+  The first trigger contains elements (1), since they arrive
+  before window closing
+  The second trigger contains elements (2) and only elements (2).
+  They are triggered by themselves as they arrived late
+  The thrid trigger contains only elements (3)
+  since they also arrived late but at different time
+
+Element "On-Time-New-Window-1" shows that this process continues
+until termination
 """

Diff for: Python/testing_windows/element_count_trigger.py

@@ -48,7 +48,8 @@ def get_input_stream():
         .advance_processing_time(2)
         .add_elements(
             [
-                # Different event times but elements "arrived" same time in pipeline
+                # Different event times but elements "arrived"
+                # same time in pipeline
                 TimestampedValue("Same-Time-1", timestamp=4),
                 TimestampedValue("Same-Time-2", timestamp=5),
                 TimestampedValue("Same-Time-3", timestamp=6),
@@ -112,18 +113,20 @@ def run():
             allowed_lateness=window_allowed_lateness_seconds,
             accumulation_mode=trigger.AccumulationMode.DISCARDING,
             # Try setting different types of triggers!
-            # AfterWatermark only fires pane once when the watermark crosses the
-            # Window end time
+            # AfterWatermark only fires pane once when the watermark
+            # crosses the Window end time
             # trigger=trigger.AfterWatermark()
-            # We  repeatedly fire panes forever with count as least late_pane_size
+            # We  repeatedly fire panes forever with count as
+            # least late_pane_size
             # We also want to set a termination to stop above infinite fires so
             # use AfterWatermark a final trigger.
             # trigger=trigger.OrFinally(
             #     trigger.Repeatedly(trigger.AfterCount(late_pane_size)),
             #     trigger.AfterWatermark()
             # )
-            # We need to configure how late data should be handled to prevent data
-            #  loss since AfterWatermark by default only fires once
+            # We need to configure how late data should be handled
+            # to prevent data loss since AfterWatermark
+            # by default only fires once
             # https://github.com/apache/beam/blob/master/sdks/python/apache_beam/transforms/trigger.py#L596
             trigger=trigger.OrFinally(
                 trigger.Repeatedly(trigger.AfterCount(late_pane_size)),
@@ -145,18 +148,27 @@ def run():
 """
 EXPLANATION
 
-`trigger.AfterCount` triggers whenever the current pane has at least `late_pane_size` elements,
- but there are some constrains.
-
-1 - The first trigger contains "First" and "Second", which is expected since they are the first 2 elements.
-2 - The second trigger contains the "Third" element but also all of "Same-Time-X" elements, so it is not just 2 elements.
-  This happens because of "Same-Time-X" arrive at the same time, so there is no second element but second elements
-3 - The third trigger only contains "Last-In-Window", this is because we have `trigger.OrFinally(... , trigger.AfterWatermark(...))`
-  so it triggers when the window closes, no matter the number of elements there are in pane.
+`trigger.AfterCount` triggers whenever the current pane has
+at least `late_pane_size` elements, but there are some constrains.
+
+1 - The first trigger contains "First" and "Second", which is expected
+    since they are the first 2 elements.
+2 - The second trigger contains the "Third" element but also
+    all of "Same-Time-X" elements, so it is not just 2 elements.
+    This happens because of "Same-Time-X" arrive at the same time,
+    so there is no second element but second elements
+3 - The third trigger only contains "Last-In-Window", this is because
+    we have `trigger.OrFinally(... , trigger.AfterWatermark(...))`
+    so it triggers when the window closes,
+    no matter the number of elements there are in pane.
 4 - The fourth trigger contains "Late-(1,2,3)", since they arrive together
-5 - The next triggers contain the elements as they come, since we have `trigger.OrFinally(... , trigger.AfterWatermark(...))`
-6 - Element "Outside-Window-Lateness" is discarded as it's outside the allowed lateness
-
-In a real life scenario, it's not that common two elements arrive at the exact same time, but networking conditions can affect arrival.
-Note: `AfterCount` can cause hanged pipeline if the condition is not met. To resolve this see note in `after_each_trigger.py`
+5 - The next triggers contain the elements as they come,
+    since we have `trigger.OrFinally(... , trigger.AfterWatermark(...))`
+6 - Element "Outside-Window-Lateness" is discarded as
+    it's outside the allowed lateness
+
+In a real life scenario, it's not that common two elements arrive
+at the exact same time, but networking conditions can affect arrival.
+Note: `AfterCount` can cause hanged pipeline if the condition is not met.
+To resolve this see note in `after_each_trigger.py`
 """

Diff for: Python/testing_windows/late_data.py

@@ -45,7 +45,8 @@ def get_input_stream():
         .advance_processing_time(3)
         .add_elements(
             [
-                # Although pane fired, below element still within lateness configured
+                # Although pane fired, below element still
+                # within lateness configured
                 TimestampedValue("Late-but-allowed", timestamp=9)
             ]
         )
@@ -90,10 +91,13 @@ def run():
         >> WindowInto(
             FixedWindows(size=window_size_seconds),
             allowed_lateness=window_allowed_lateness_seconds,
-            # Set `accumulation_mode` as `ACCUMULATING` when you want to collect all the
-            # panes fired in the window. To have the panes be considered individually
-            # use `DISCARDING` mode. Use `DISCARDING` when the datum are indipendent
-            # of each other. `ACCUMULATING` will fire each time the a panes arrives from a window
+            # Set `accumulation_mode` as `ACCUMULATING` when you want to
+            # collect all the panes fired in the window.
+            # To have the panes be considered individually
+            # use `DISCARDING` mode. Use `DISCARDING`
+            # when the datum are independent of each other.
+            # `ACCUMULATING` will fire each time
+            # when a panes arrives from a window
             accumulation_mode=AccumulationMode.DISCARDING,
         )
         | "Group together elements within same pane and key" >> GroupByKey()
@@ -111,14 +115,21 @@ def run():
 
 #   EXPLANATION
 # 1 - Elements A,B,C,D arrive to the pipeline on time
-# 1 - Element "Late-but-before-window-closing" arrives 9s late, but before window closes
-# 2 - Element "Late-but-allowed" arrives after window closing, but within the allowed late time
-# 3 - Element "Late-and-outside-window" arrives after window closing and after allowed late time
-# 4 - Element "window-2" arrives on time but in a different window than 0 to 10000
+# 1 - Element "Late-but-before-window-closing" arrives 9s late,
+#     but before window closes
+# 2 - Element "Late-but-allowed" arrives after window closing,
+#     but within the allowed late time
+# 3 - Element "Late-and-outside-window" arrives after window closing and
+#     after allowed late time
+# 4 - Element "window-2" arrives on time but in a different window
+#     than 0 to 10000
 #
-# The first trigger with elements (1) contains all data from the first window that arrived before window closed
-# The second trigger with element (2) includes the data that arrived late but within the allowed late time
+# The first trigger with elements (1) contains all data from the first window
+# that arrived before window closed
+# The second trigger with element (2) includes the data that arrived late
+# but within the allowed late time
 # Elements in (3) are discarded, since they arrive after the allowed late time
-# The third trigger with element (4) appears in a new window `[10.0, 20.0)` with pane index `0`
-# Even though the Window is same for (1) (2) the pane index for (1) is `0` and for (2) is `1`
-# denoting they are logically separated
+# The third trigger with element (4) appears in a new window `[10.0, 20.0)`
+# with pane index `0`
+# Even though the Window is same for (1) (2) the pane index for (1) is `0`
+# and for (2) is `1` denoting they are logically separated