Simple example of using pexpect+asyncio from non async code

simple-asyncio.py

# Use it with: python3 simple-asyncio.py <timeout>
# timeout defaults to 5 seconds which means it'll never timeout
# change the string to expect (or the string printed after the sleep) to demonstrate the EOF example (did not timeout but did not match)

import sys
import asyncio
import pexpect


async def work(timeout):

    print("Starting subprocess...")
    child = pexpect.spawn("/bin/bash", ["-c", "sleep 4 && echo done"])

    async def expecter():
        # child.expect(..., async_=True) returns an Awaitable[int] which cannot be directly
        # turned into an asyncio.Task.
        # However, if you await on it from an async def, that wrapper function *can* be turned into
        # a task.
        return await child.expect(
            ["done", pexpect.EOF, pexpect.TIMEOUT], async_=True, timeout=timeout
        )

    doneT = asyncio.create_task(expecter())

    # now we check on `doneT` every 1 second or we print a spinner
    while not doneT.done():
        await asyncio.sleep(1)
        print(".", end="", flush=True)

    print()
    print("all done")

    r = doneT.result()
    if r == 0:
        print("completed successfully")
    elif r == 1:
        print("encountered eof")
    elif r == 2:
        print("timed out")

    # child.after changes depending on the index of the matched expectation
    # for a timeout or EOF, child.after is the exception class
    # for a match, child.after is the string that matched
    # note that the returned index lines up with the list elements supplied to `child.expect`
    print("after:", child.after)

    # child.match is an re.Match object if there was a string match, otherwise it's the same as
    # child.after
    print("match:", child.match)


def inner(timeout):
    return asyncio.run(work(timeout))


def main(timeout):
    print("Starting work with a timeout of", timeout)
    return inner(timeout)


if __name__ == "__main__":
    timeout = 5
    if len(sys.argv) > 1:
        timeout = int(sys.argv[1])
    main(timeout)