Check the order of results

Question

Summary Append to list only if that value is the next value expected.

"""
Return codes can be wrong because errors happen. And the next return code can be reliant on this one.

This module checks that a return code is expected next and that only then does it become the latest piece of a
 fulfilling order.
"""

__all__ = ['order_two_with_retry', 'wait_on_the_return_of_a_commandline']

import functools
import operator
import subprocess
import sys
import time
import typing

insert_object_at_the_end_of_the_list = list.append


# optimization
def if_the_list_is_empty(insertion: typing.Any, out: typing.List, expectation: typing.List) -> None:
    assert not out, 'the list is not empty'

    a = [insertion]

    if a == expectation[:1]:
        insert_object_at_the_end_of_the_list(out, insertion)


def if_the_list_is_not_empty(insertion: typing.Any, out: typing.List, expectation: typing.List) -> None:
    a = out[::]
    insert_object_at_the_end_of_the_list(a, insertion)

    if a == expectation[:len(a)]:
        insert_object_at_the_end_of_the_list(out, insertion)
    else:
        out.clear()


def lever(insertion: typing.Any, out: typing.List, expectation: typing.List) -> None:
    """Test equality (one)."""
    if not out:
        if_the_list_is_empty(insertion, out, expectation)
    else:
        if_the_list_is_not_empty(insertion, out, expectation)


def order_two(insertionm: typing.Callable, insertionn: typing.Callable, out: typing.List,
              expectation: typing.List) -> None:
    lever(insertionm(), out, expectation)
    print(out)

    if not out:
        return

    lever(insertionn(), out, expectation)
    print(out)


def order_two_with_retry(insertionm: typing.Callable, insertionn: typing.Callable, out: typing.List,
                         expectation: typing.List) -> None:
    order_two(insertionm, insertionn, out, expectation)

    if not out:
        order_two(insertionm, insertionn, out, expectation)


# Test equality (all).
test_insertions_against_expectation = operator.eq


def main():
    insertionm = functools.partial(wait_on_the_return_of_a_commandline,
                                   'python -c "import random; import sys; sys.exit(random.randrange(0, 1 + 1))"')
    insertionn = insertionm
    out = []
    expectation = [0, 0]

    order_two_with_retry(insertionm, insertionn, out, expectation)


# This function can silence this module.
# Reestablish noise with builtins.print
def set_print(new_print):
    # noinspection PyShadowingBuiltins,PyGlobalUndefined
    global print

    # noinspection PyShadowingBuiltins
    print = new_print


def filler(executingprocess):
    s = 0.4

    while executingprocess.poll() is None:
        sys.stdout.write('\r.  ')

        time.sleep(s)
        sys.stdout.write('\r.. ')

        time.sleep(s)
        sys.stdout.write('\r...')

        time.sleep(s)


def wait_on_the_return_of_a_commandline(commandline):
    executingprocess = subprocess.Popen(commandline)

    filler(executingprocess)
    return executingprocess.returncode


if __name__ == '__main__':
    main()

Comments

• Dunder all contains the two functions necessary to get the module running.
• In if_the_list... the new list is named a because a, b are the names of arguments to operators and I'm testing for equality (operator.eq(a, b)).
• The comment 'optimization' is there because that function does less steps then the next function for the same computation.
• out[::] is faster than list(out) on my computer; use of slices not to modify lists too soon and by accident (slices create new lists).
• If this returncode does not match the expected returncode, out is cleared because I want to try again, not ignore and keep going.
• order_two_with_retry retries once, and if I have an odd number, I should use lever for that. This is where the return code appears by calling a function or opening a subprocess.
• The alias to operator.eq: the name fits it into the module. Test for equality: out is a, expectation is b.
• main generates 0 or 1 from a subprocess randomly. It shows module usage.
• prints in order_two can be silenced by setting print to something like def print(*args, **kwargs): pass
• And finally, wait on subprocess exitcode with a console animation that repeats one dot, two, three.

Question

I'm looking for tips on commenting code as I'm not sure how to do it or what the reader needs.

Answer 1

The comment 'optimization' is there because that function does less steps then the next function for the same computation.

I'm willing to bet that this has no impact at all, and even if it did it's premature optimisation. The same applies to out[::] is faster than list(out) on my computer. Think: you're calling a child process. That will overwhelmingly be the bottleneck, right?

Meant in the nicest way possible, most of this code is insane and bad and needs to go away. Names like insert_object_at_the_end_of_the_list are gratuitously long and do not help. if_the_list_is_empty does not at all describe what the function actually does, which is to insert only if the expected prefix is seen. But this function should not exist.

Don't assert in production code.

lever is an equally unhelpful function name.

Your use of subprocess assumes shell=True which is bad. Avoid this if possible. You only show a toy example so it's impossible to provide more advice on what you're actually doing.

Delete set_print. It's unused, you don't show how it would be used, and I see no utility in it.

poll() is not the correct Popen call to make here; instead call wait passing in your polling interval as a timeout. Importantly, this will bail early once the process is done, where your sleep will not. Do not sleep.

Put your Popen instance in context management.

Don't use random numbers in what should be a deterministic test.

Many consoles won't output anything from a print() unless it terminates in a newline or has a flush(). An alternative to stdout.write is a print with its end='\r' followed by a flush. You should be putting the carriage return at the end and not the beginning of your string, to allow the next line of "real" output to overwrite it once the animation is done. Also, the animation will only be visible for the first three cycles unless you pad with spaces.

Somewhat reading into what you "actually want", let's assume that you're trying to make a generic Popen poll-and-animate function, and a separate retry-on-failure function. The latter is easy as a parametric decorator:

Suggested

from subprocess import Popen, TimeoutExpired, SubprocessError
from sys import stdout
from typing import Callable, Any


def wait_process(command: str, poll_interval: float = 0.4, expected: int = 0) -> None:
    with Popen(command) as process:
        while True:
            for dots in range(1, 4):
                try:
                    process.wait(timeout=poll_interval)
                    if process.returncode == expected:
                        return
                    raise SubprocessError(
                        f'"{command}" completed with code {process.returncode}, '
                        f'expected {expected}'
                    )
                except TimeoutExpired:
                    pass

                print(f'{"."*dots:3}', end='\r')
                stdout.flush()


def retry_process(times: int) -> Callable:
    def decorator(inner: Callable) -> Callable:
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            for attempt in range(times - 1):
                try:
                    return inner(*args, **kwargs)
                except SubprocessError:
                    pass

            return inner(*args, **kwargs)
        return wrapper
    return decorator


@retry_process(times=2)
def main() -> None:
    print('Starting first')
    command = 'python -c "from time import sleep; sleep(2); exit(0)"'
    wait_process(command=command, expected=0)
    print('First completed, starting second')
    wait_process(command=command, expected=1)
    print('Second completed')


if __name__ == '__main__':
    main()

Output

Starting first
.
..
...
.
..
First completed, starting second
.
..
...
.
..
Starting first
.
..
...
.
..
First completed, starting second
.
..
...
.
..
Traceback (most recent call last):
  File "C:\Users\gtoom\src\stackexchange\276074.py", line 47, in <module>
    main()
  File "C:\Users\gtoom\src\stackexchange\276074.py", line 32, in wrapper
    return inner(*args, **kwargs)
  File "C:\Users\gtoom\src\stackexchange\276074.py", line 42, in main
    wait_process(command=command, expected=1)
  File "C:\Users\gtoom\src\stackexchange\276074.py", line 13, in wait_process
    raise SubprocessError(
subprocess.SubprocessError: "python -c "from time import sleep; sleep(2); exit(0)"" completed with code 0, expected 1