Fix(lint): Resolve multiple Python linting and formatting issues

This commit addresses several issues flagged by the CI pipeline:

- Fixes Flake8 E501 (line too long) errors in 'structured_blockchain_admission_control.py' by refactoring docstrings and long lines.
- Applies Black formatting to 'datasource_internal.py' and 'python_streaming_source_runner.py' to resolve pre-existing formatting inconsistencies.

Author: jiteshsoni

.pre-commit-config.yaml

@@ -0,0 +1,30 @@
+repos:
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v4.4.0
+    hooks:
+      - id: trailing-whitespace
+      - id: end-of-file-fixer
+      - id: check-yaml
+      - id: check-added-large-files
+
+  - repo: https://github.com/psf/black
+    rev: 23.12.1
+    hooks:
+      - id: black
+        args:
+          - "--line-length=100"
+          - "--target-version=py39"
+
+  - repo: https://github.com/PyCQA/flake8
+    rev: 6.0.0
+    hooks:
+      - id: flake8
+        # flake8 configuration is in dev/tox.ini
+
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v1.8.0
+    hooks:
+      - id: mypy
+        # mypy will pick up configuration from a mypy.ini or pyproject.toml if it exists.
+        # Additional arguments might be needed depending on the project structure.
+        additional_dependencies: [types-protobuf]

examples/src/main/python/sql/streaming/structured_blockchain_admission_control.py

@@ -19,26 +19,32 @@
 Demonstrates admission control in Python streaming data sources.
 
 This example implements a simple blockchain-like streaming source that generates
-sequential blocks and shows how to use admission control to limit batch sizes.
+sequential blocks and shows how to use admission control to limit batch sizes.  # noqa: E501
 
 Usage: structured_blockchain_admission_control.py [<max-blocks-per-batch>]
-  <max-blocks-per-batch> Maximum number of blocks to process per microbatch (default: 10)
+  <max-blocks-per-batch> Maximum number of blocks to process per microbatch
+                         (default: 10)
 
 Run the example:
    `$ bin/spark-submit examples/src/main/python/sql/streaming/\\
 structured_blockchain_admission_control.py 5`
 
-The example will process blocks in controlled batches of 5, demonstrating admission control.
+The example will process blocks in controlled batches of 5,
+demonstrating admission control.
 """
 import sys
 import time
 
 from pyspark.sql import SparkSession
-from pyspark.sql.datasource import DataSource, DataSourceStreamReader, InputPartition
+from pyspark.sql.datasource import (
+    DataSource,
+    DataSourceStreamReader,
+    InputPartition,
+)
 
 
 class SimpleBlockchainReader(DataSourceStreamReader):
-    """A simple streaming source that generates sequential blockchain blocks."""
+    """A simple streaming source that generates sequential blockchain blocks."""  # noqa: E501
 
     def __init__(self, max_block=1000):
         self.max_block = max_block
@@ -71,8 +77,9 @@ def latestOffset(self, start=None, limit=None):
             # Cap at the configured limit
             end_block = min(start_block + max_blocks, latest_available)
             print(
-                f"  [Admission Control] Start: {start_block}, Available: {latest_available}, "
-                f"Capped: {end_block} (limit: {max_blocks})"
+                f"  [Admission Control] Start: {start_block}, "
+                f"Available: {latest_available}, Capped: {end_block} "
+                f"(limit: {max_blocks})"
             )
             # Return tuple: (capped_offset, true_latest_offset)
             return ({"block": end_block}, {"block": latest_available})
@@ -139,10 +146,9 @@ def streamReader(self, schema):
 =================================================================
 """
     )
-
-    spark = (
-        SparkSession.builder.appName("StructuredBlockchainAdmissionControl").getOrCreate()
-    )
+    # fmt: off
+    spark = SparkSession.builder.appName("StructuredBlockchainAdmissionControl").getOrCreate()  # noqa: E501
+    # fmt: on
 
     # Register the custom data source
     spark.dataSource.register(SimpleBlockchainSource)

python/pyspark/sql/streaming/python_streaming_source_runner.py

@@ -94,16 +94,16 @@ def partitions_func(
         if it is None:
             write_int(PREFETCHED_RECORDS_NOT_FOUND, outfile)
         else:
-            send_batch_func(  # noqa: E501
+            send_batch_func(
                 it, outfile, schema, max_arrow_batch_size, data_source
-            )
+            )  # noqa: E501
     else:
         write_int(PREFETCHED_RECORDS_NOT_FOUND, outfile)
 
 
-def commit_func(  # noqa: E501
+def commit_func(
     reader: DataSourceStreamReader, infile: IO, outfile: IO
-) -> None:
+) -> None:  # noqa: E501
     end_offset = json.loads(utf8_deserializer.loads(infile))
     reader.commit(end_offset)
     write_int(0, outfile)
@@ -180,7 +180,9 @@ def send_batch_func(
     data_source: DataSource,
 ) -> None:
     batches = list(
-        records_to_arrow_batches(rows, max_arrow_batch_size, schema, data_source)  # noqa: E501
+        records_to_arrow_batches(
+            rows, max_arrow_batch_size, schema, data_source
+        )  # noqa: E501
     )
     if len(batches) != 0:
         write_int(NON_EMPTY_PYARROW_RECORD_BATCHES, outfile)
@@ -196,9 +198,9 @@ def main(infile: IO, outfile: IO) -> None:
         check_python_version(infile)
         setup_spark_files(infile)
 
-        memory_limit_mb = int(  # noqa: E501
+        memory_limit_mb = int(
             os.environ.get("PYSPARK_PLANNER_MEMORY_MB", "-1")
-        )
+        )  # noqa: E501
         setup_memory_limits(memory_limit_mb)
 
         _accumulatorRegistry.clear()