Skip to main content

Asyncio implementation of Cmd Python lib.

Project description

Build Status codecov PyPI version PyPI

asynccmd

Async implementation of Cmd Python lib.

Asynccmd is a library to build command line interface for you asyncio project.

It’s very simple like original Cmd lib https://docs.python.org/3.6/library/cmd.html.

The mechanic is very similar. You have Cmd superclass, you can override class method and add yours own.

Features

  • support command line for Windows and POSIX systems

  • build-in help or ? command to list all available command

  • build-in exit command for emergency stop asyncio loop

  • support repeat last cmd command by sending empty string

Getting started

Simple example

This is very simple example to show you main features and how they can be used.

First of all, we are create new class and inherited our Cmd class. Do not instantiate Cmd itself.

Than create instance of this new class and run loop.

class SimpleCommander(Cmd):
    def __init__(self, mode, intro, prompt):
        # We need to pass in Cmd class mode of async cmd running
        super().__init__(mode=mode)
        self.intro = intro
        self.prompt = prompt
        self.loop = None

    def do_tasks(self, arg):
        """
        Our example method. Type "tasks <arg>"
        :param arg: contain args that go after command
        :return: None
        """
        for task in asyncio.Task.all_tasks(loop=self.loop):
            print(task)

    def start(self, loop=None):
        # We pass our loop to Cmd class.
        # If None it try to get default asyncio loop.
        self.loop = loop
        # Create async tasks to run in loop. There is run_loop=false by default
        super().cmdloop(loop)

# For win system we have only Run mode
# For POSIX system Reader mode is preferred


if sys.platform == 'win32':
    loop = asyncio.ProactorEventLoop()
    mode = "Run"
else:
    loop = asyncio.get_event_loop()
    mode = "Reader"
# create instance
cmd = SimpleCommander(mode=mode, intro="This is example", prompt="example> ")
cmd.start(loop)  # prepaire instance
try:
    loop.run_forever()  # our cmd will run automatilly from this moment
except KeyboardInterrupt:
    loop.stop()

Link to simple.py

General example

We use our simple example, but add some new staff: * sleep_n_print coroutine that will be called from our cli command * do_sleep new method (sleep cli command) that add task to event loop

async def sleep_n_print(loop, time_to_sleep=None):
    """
    This is our simple coroutine.
    :param time_to_sleep: time to sleep in seconds
    :return: await sleep for time_to_sleep seconds
    """
    asyncio.set_event_loop(loop)  # set correct event loop
    await asyncio.sleep(int(time_to_sleep))
    print("Wake up! I was slept for {0}s".format(time_to_sleep))
def do_sleep(self, arg):
    """
    Our example cmd-command-method for sleep. sleep <arg>
    :param arg: contain args that go after command
    :return: None
    """
    self.loop.create_task(sleep_n_print(self.loop, arg))

Link to main.py

Run our cli and make sleep 10 command 3 times. Now we have 3 sleep_n_print async tasks in our event loop. If you use tasks command, you see something like that.

example>tasks
<Task pending coro=<sleep_n_print() running at asynccmd\examples\main.py:13> wait_for=<Future pending cb=[Task._wakeup()]>>
<Task pending coro=<Cmd._read_line() running at C:\Program Files\Python35\lib\site-packages\asynccmd\asynccmd.py:141>>
<Task pending coro=<sleep_n_print() running at asynccmd\examples\main.py:13> wait_for=<Future pending cb=[Task._wakeup()]>>
<Task pending coro=<sleep_n_print() running at asynccmd\examples\main.py:13> wait_for=<Future pending cb=[Task._wakeup()]>>
example>
Wake up! I was slept for 10s
Wake up! I was slept for 10s
Wake up! I was slept for 10s

Aiohttp implementation

This is practical example how to control aiohttp instances. We will create two cli command start and stop. This commands get port number as only one argument. Let’s make some changes for our general example:

Create class helper that will be do all aiohttp staff for us.

class AiohttpCmdHelper:
    """
    Helper class that do all aiohttp start stop manipulation
    """
    port = 8080  # Default port
    loop = None  # By default loop is not set

    def __init__(self, loop, port):
        self.loop = loop
        self.port = port

    async def handle(self, request):
        """
        Simple handler that answer http request get with port and name
        """
        name = request.match_info.get('name', "Anonymous")
        text = 'Aiohttp server running on {0} port. Hello, {1}'.format(
            str(self.port), str(name))
        return web.Response(text=text)

    async def start(self):
        """
        Start aiohttp web server
        """
        self.app = web.Application()
        self.app.router.add_get('/', self.handle)
        self.app.router.add_get('/{name}', self.handle)
        self.handler = self.app.make_handler()
        self.f = self.loop.create_server(self.handler,
                                         host='0.0.0.0',
                                         port=self.port)
        # Event loop is already runing, so we await create server instead
        # of run_until_complete
        self.srv = await self.f

    async def stop(self):
        """
        Stop aiohttp server
        """
        self.srv.close()
        await self.srv.wait_closed()
        await self.app.shutdown()
        await self.handler.shutdown(60.0)
        await self.app.cleanup()

Now we ready to add start and stop command to Commander.

# Add property to store helper objects
    aiohttp_servers = []
# ...

def do_start(self, arg):
    """
    Our example cli-command-method for start aiohttp server. start <arg>
    :param arg: Port number
    :return: None
    """
    if not arg:  # we use simple check in our demonstration
        print("Error port is empty")
    else:
        test = AiohttpCmdHelper(loop=self.loop, port=int(arg))
        self.aiohttp_servers.append({'port': int(arg),'server': test})
        self.loop.create_task(test.start())

def do_stop(self, arg):
    """
    Our example cli-command-method for stop aiohttp server. start <arg>
    :param arg: Port number
    :return: None
    """
    if not arg:  # we use simple check in our demonstration
        print("Error! Provided port is empty")
    else:
        aiohttp_servers = []
        for srv in self.aiohttp_servers:
            if srv['port'] == int(arg):
                self.loop.create_task(srv['server'].stop())
            else:
                aiohttp_servers.append({'port': srv['port'], 'server': srv['server']})
        self.aiohttp_servers = aiohttp_servers

We need to add asyncio.set_event_loop(loop) addition to our main example to prevent aiohttp to create its own loop.

if sys.platform == 'win32':
    loop = asyncio.ProactorEventLoop()
    mode = "Run"
else:
    loop = asyncio.get_event_loop()
    mode = "Reader"

asyncio.set_event_loop(loop)  # set our event loop for aiohttp (fix for Win32)

That’s all. Now we can run multiple aiohttp server from our code.

Link to aiohttp_example.py

Documentation

TBD

Contributing

Main stream is fork project, commit changes and send pull request. Contributing to lib you could make in form of feedback, bug reports or pull requests. CONTRIBUTING.md - TBD.

Requirements

  • Python >= 3.5

License

asynccmd is offered under the Apache 2 license.

Source code

The latest developer version is available at https://github.com/valentinmk/asynccmd

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

asynccmd-0.2.4.tar.gz (7.0 kB view details)

Uploaded Source

Built Distributions

asynccmd-0.2.4-py3.7.egg (9.6 kB view details)

Uploaded Source

asynccmd-0.2.4-py3.6.egg (9.5 kB view details)

Uploaded Source

asynccmd-0.2.4-py3.5.egg (9.6 kB view details)

Uploaded Source

File details

Details for the file asynccmd-0.2.4.tar.gz.

File metadata

  • Download URL: asynccmd-0.2.4.tar.gz
  • Upload date:
  • Size: 7.0 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No

File hashes

Hashes for asynccmd-0.2.4.tar.gz
Algorithm Hash digest
SHA256 bb4d0a91e063b51e5c61c79e2d551485f83c5cbfa85cd868814efe6fa494b8b2
MD5 9362ab63b4260682007631e20aac239f
BLAKE2b-256 e2cc9c1d38d4e8f77aa4b0fc6350dc28fb045249cbc540558255dcf9f0c880f4

See more details on using hashes here.

File details

Details for the file asynccmd-0.2.4-py3.7.egg.

File metadata

File hashes

Hashes for asynccmd-0.2.4-py3.7.egg
Algorithm Hash digest
SHA256 34855d68f0c069bacf810586ef4e87764ca97bdb3c72a1961db5528aaf608070
MD5 550942b3655092348dda88e18c8b2a44
BLAKE2b-256 508102c9ee0ad2d0db5564c4b051c2d7531dc57cbb9b8be472c29b0c4b55812a

See more details on using hashes here.

File details

Details for the file asynccmd-0.2.4-py3.6.egg.

File metadata

File hashes

Hashes for asynccmd-0.2.4-py3.6.egg
Algorithm Hash digest
SHA256 1661e295cddd666f532c10f2d1ad99fcaaa6258e50a48a15ea7dfce7bd10065c
MD5 a7085296137ba88997210c269dfba818
BLAKE2b-256 ec4da04186e8145630124fbb6af5c50e10ae8f0f69d2658b546929b51027a28b

See more details on using hashes here.

File details

Details for the file asynccmd-0.2.4-py3.5.egg.

File metadata

File hashes

Hashes for asynccmd-0.2.4-py3.5.egg
Algorithm Hash digest
SHA256 b59d59bd4fdc4feb46b886ec8031f0a968145ff640fde27909da138b7974e13e
MD5 88451bb45fe049c774b5de2f2b40543c
BLAKE2b-256 adb714825bf430c0e333d44137bb8a07aa2ff220e8a3734632358ddb925cc33b

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page