Mixing cli and repl user interface in a sensible way, with extensive unit testing.

click_repl_cmd.py

import hypothesis
from pydantic import BaseModel, ValidationError, validate_model
import unittest


class Answer(BaseModel):

    # Data model of user interface (might be different than data model of Exchange interface)

    data: int

    def __init__(self, data: int):
        super().__init__(data=data)

    def printme(self):
        print(self.data)
        # Complicated code will be here
        return True


class TestAnswer(unittest.TestCase):

    # NOTE : later we can extract complex hypothesis test structures with pydantic into separate packages
    # already done for other parsers : https://hypothesis.readthedocs.io/en/latest/strategies.html#external-strategies

    @hypothesis.settings(verbosity=hypothesis.Verbosity.verbose)
    @hypothesis.given(a=hypothesis.infer)
    def test_printme(self, a: int):

        answer = Answer(a)
        assert answer.printme()




import prompt_toolkit

from prompt_toolkit.eventloop.defaults import use_asyncio_event_loop
use_asyncio_event_loop()

from prompt_toolkit.patch_stdout import patch_stdout


# EVENTUALLY:
#
# from prompt_toolkit.application import Application
#
# async def repl():
#     # Define application.
#     application = Application(
#         ...
#     )
#
#     result = await application.run_async()
#     print(result)
#
# asyncio.get_event_loop().run_until_complete(main())

async def input_answer() -> Answer:
    a = None
    with patch_stdout():
        data = await prompt_toolkit.prompt(message="Enter Answer: ", async_=True)
        while a is None:
            values, errors = validate_model(model=Answer, input_data={'data': data}, raise_exc=False)
            if errors:
                print(errors)
                data = await prompt_toolkit.prompt(message="Enter Answer AGAIN: ", async_=True)
            else:
                a = Answer(**values)
    return a


counter = [0]


async def print_counter():
    """
    Coroutine that prints counters and saves it in a global variable.
    """
    import asyncio

    while True:
        print('Counter: %i' % counter[0])
        counter[0] += 1
        await asyncio.sleep(3)


def repl():
    import asyncio
    loop = asyncio.get_event_loop()

    import asyncio
    counter = asyncio.ensure_future(print_counter())
    result = asyncio.ensure_future(input_answer())

    def stopcount(finished_task):
        counter.cancel()

    result.add_done_callback(stopcount)

    loop.run_until_complete(asyncio.gather(counter, result))

    result.result().printme()
    loop.close()


if __name__ == '__main__':

    import click

    CONTEXT_SETTINGS = dict(help_option_names=['-h', '--h', '--help', '?'])

    @click.group(context_settings=CONTEXT_SETTINGS, invoke_without_command=False)
    @click.pass_context
    def cli_group_example(ctx):
        click.echo(f"Ask stuff")

    @cli_group_example.command('ask')
    @click.option('-a', '--answer', type=int, default=42, show_default=True,
                  help='comma delimited list of assets to restrict output to')
    @click.pass_obj
    def click_cmd_exmple(obj, answer):
        a = Answer(answer)
        a.printme()

        return 0  # shell success

    @cli_group_example.command('repl')
    @click.pass_obj
    def cli_repl(obj):
        repl()
        return 0  # shell success


    @cli_group_example.command('tryme')
    @click.pass_obj
    def tryme(obj):
        # initialize the test suite
        loader = unittest.TestLoader()
        suite = unittest.TestSuite()

        # add tests to the test suite
        suite.addTests(loader.loadTestsFromTestCase(TestAnswer))

        # initialize a runner, pass it your suite and run it
        runner = unittest.TextTestRunner(verbosity=3)
        return runner.run(suite)

    cli_group_example()