Support for line-based streaming of subprocess output

[asnare]

This PR factors out some common code (from databrickslabs/lakebridge#2163) needed for processing a line-based stream in realtime, such as when working with the output of subprocesses. This is currently needed by the Lakebridge CLI as well as the BB LSP Server.

Although moderately complex, this is needed because the builtin Python support for line-based processing¹ lacks in several ways:

• When a line exceeds the limit it throws an exception. Exceptions in asynchronous contexts are very hard to deal with, making it error-prone. If unhandled (eg. in a task) everything stalls.
• When an error occurs there's no reasonable recovery path: the internal and stream state are basically unknown.

In contrast, this processing:

• Handles lines of arbitrary length, splitting up long lines if necessary.
• Uses bounded memory irrespective of how long the lines are; unnecessarily large buffers are not needed.

Expected usage is something like:

async def run_something() -> None:
    process = await asyncio.create_subprocess_exec(
        ...,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    assert process.stdout is not None, "stdout should be a pipe"
    assert process.stderr is not None, "stderr should be a pipe"

    async def log_stream(name: str, stream: asyncio.StreamReader, log_level: int, limit: int = 4096) -> None:
        try:
            async for line in readlines(stream=stream, limit=limit):
                logger.log(log_level, str(line))
        except Exception as e:
            logger.error(f"Error reading {name}: {e}", exc_info=True)
            # Drain to prevent blocking.
            try:
                while await stream.read(limit): pass
            except Exception:
                pass  # Exception while draining, situation seems unrecoverable.

    await asyncio.gather(
        log_stream("stdout", process.stdout, logging.DEBUG),
        log_stream("stderr", process.stderr, logging.INFO),
        process.wait(),
    )

    return process.returncode

Footnotes

1. asyncio.StreamReader.readline() ↩

[github-actions]

✅ 40/40 passed, 2 skipped, 1m44s total

_{Running from acceptance #376}

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 80.48%. Comparing base (7ebe9a3) to head (e79b8dd).
⚠️ Report is 2 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #320      +/-   ##
==========================================
+ Coverage   80.05%   80.48%   +0.43%     
==========================================
  Files          17       17              
  Lines        2396     2449      +53     
  Branches      516      526      +10     
==========================================
+ Hits         1918     1971      +53     
  Misses        354      354              
  Partials      124      124              

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[m-abulazm]

Nice work. I just have a couple of questions to make sure I understand everything.