Add Async Support

.gitignore

@@ -26,3 +26,5 @@ MANIFEST
 .vscode
 
 Pipfile.lock
+
+.venv

README.rst

@@ -3,3 +3,14 @@ Documentation
 
 Documentation resides over at
 `readthedocs <https://librouteros.readthedocs.io/>`_
+
+Async Usage
+connection = await async_connect(
+        host="127.0.0.1",
+        username="admin",
+        password="admin",
+        timeout=30
+    )
+
+hotspot_user_path = connection.path('/ip/hotspot/user')
+hotspot_users = [user async for user in hotspot_user_path.select('.id')]

apicli.py

@@ -3,7 +3,6 @@
 
 """Command line interface for debugging purpouses."""
 
-
 import logging
 import getpass
 from argparse import ArgumentParser
@@ -14,19 +13,16 @@
 from librouteros import connect
 from librouteros.exceptions import TrapError, FatalError
 
-argParser = ArgumentParser(description='mikrotik api cli interface')
-argParser.add_argument(
-        'host', type=str,
-        help="host to with to connect. may be fqdn, ipv4 or ipv6 address")
-argParser.add_argument('-u', '--user', type=str, required=True, help="username")
-argParser.add_argument(
-        '-p', '--port', type=int, default=8728, help="port to connect to (default 8728)")
+argParser = ArgumentParser(description="mikrotik api cli interface")
+argParser.add_argument("host", type=str, help="host to with to connect. may be fqdn, ipv4 or ipv6 address")
+argParser.add_argument("-u", "--user", type=str, required=True, help="username")
+argParser.add_argument("-p", "--port", type=int, default=8728, help="port to connect to (default 8728)")
 args = argParser.parse_args()
 
-mainlog = logging.getLogger('librouteros')
+mainlog = logging.getLogger("librouteros")
 console = logging.StreamHandler(stdout)
 mainlog.setLevel(logging.DEBUG)
-formatter = logging.Formatter(fmt='%(message)s')
+formatter = logging.Formatter(fmt="%(message)s")
 console.setFormatter(formatter)
 mainlog.addHandler(console)
 
@@ -72,5 +68,5 @@ def main():
             api.close()
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     main()

librouteros/__init__.py

@@ -1,21 +1,24 @@
 # -*- coding: UTF-8 -*-
 
+import asyncio
+
 from socket import create_connection
 from collections import ChainMap
 
 from librouteros.exceptions import (
     ConnectionClosed,
     FatalError,
 )
-from librouteros.connections import SocketTransport
-from librouteros.protocol import ApiProtocol
+from librouteros.connections import SocketTransport, AsyncSocketTransport
+from librouteros.protocol import ApiProtocol, AsyncApiProtocol
 from librouteros.login import (
     plain,
     token,  # noqa: F401 BACK_COMP
+    async_plain,
 )
-from librouteros.api import Api
+from librouteros.api import Api, AsyncApi
 
-DEFAULTS = {
+SYNC_DEFAULTS = {
     "timeout": 10,
     "port": 8728,
     "saddr": "",
@@ -25,6 +28,16 @@
     "login_method": plain,
 }
 
+ASYNC_DEFAULTS = {
+    "timeout": 10,
+    "port": 8728,
+    "saddr": "",
+    "subclass": AsyncApi,
+    "encoding": "ASCII",
+    "ssl_wrapper": lambda sock: sock,
+    "login_method": async_plain,
+}
+
 
 def connect(host: str, username: str, password: str, **kwargs) -> Api:
     """
@@ -41,7 +54,7 @@ def connect(host: str, username: str, password: str, **kwargs) -> Api:
     :param ssl_wrapper: Callable (e.g. ssl.SSLContext instance) to wrap socket with.
     :param login_method: Callable with login method.
     """
-    arguments = ChainMap(kwargs, DEFAULTS)
+    arguments = ChainMap(kwargs, SYNC_DEFAULTS)
     transport = create_transport(host, **arguments)
     protocol = ApiProtocol(transport=transport, encoding=arguments["encoding"])
     api: Api = arguments["subclass"](protocol=protocol)
@@ -54,7 +67,43 @@ def connect(host: str, username: str, password: str, **kwargs) -> Api:
         raise
 
 
+async def async_connect(host: str, username: str, password: str, **kwargs) -> AsyncApi:
+    """
+    Connect and login to routeros device.
+    Upon success return a Api class.
+
+    :param host: Hostname to connecto to. May be ipv4,ipv6,FQDN.
+    :param username: Username to login with.
+    :param password: Password to login with. Only ASCII characters allowed.
+    :param timeout: Socket timeout. Defaults to 10.
+    :param port: Destination port to be used. Defaults to 8728.
+    :param saddr: Source address to bind to.
+    :param subclass: Subclass of Api class. Defaults to Api class from library.
+    :param ssl_wrapper: Callable (e.g. ssl.SSLContext instance) to wrap socket with.
+    :param login_method: Callable with login method.
+    """
+    arguments = ChainMap(kwargs, ASYNC_DEFAULTS)
+    transport = await async_create_transport(host, **arguments)
+    protocol = AsyncApiProtocol(transport=transport, encoding=arguments["encoding"], timeout=arguments["timeout"])
+    api: AsyncApi = arguments["subclass"](protocol=protocol)
+
+    try:
+        await arguments["login_method"](api=api, username=username, password=password)
+        return api
+    except (ConnectionClosed, FatalError):
+        await transport.close()
+        raise
+
+
 def create_transport(host: str, **kwargs) -> SocketTransport:
     sock = create_connection((host, kwargs["port"]), kwargs["timeout"], (kwargs["saddr"], 0))
     sock = kwargs["ssl_wrapper"](sock)
     return SocketTransport(sock=sock)
+
+
+async def async_create_transport(host: str, **kwargs) -> AsyncSocketTransport:
+    reader, writer = await asyncio.wait_for(
+        asyncio.open_connection(host=host, port=kwargs["port"]),
+        timeout=kwargs["timeout"],
+    )
+    return AsyncSocketTransport(reader=reader, writer=writer)

librouteros/api.py

@@ -7,12 +7,14 @@
     compose_word,
     parse_word,
     ApiProtocol,
+    AsyncApiProtocol,
 )
 from librouteros import query
 
 from librouteros.types import (
     ReplyDict,
     ResponseIter,
+    AsyncResponseIter,
     Response,
 )
 
@@ -140,3 +142,138 @@ def update(self, **kwargs: typing.Any) -> None:
                 **kwargs,
             )
         )
+
+
+class AsyncApi:
+    def __init__(self, protocol: AsyncApiProtocol):
+        self.protocol = protocol
+
+    async def __call__(self, cmd: str, **kwargs: typing.Any) -> AsyncResponseIter:
+        """
+        Call Api with given command.
+        Yield each row.
+
+        :param cmd: Command word. eg. /ip/address/print
+        :param kwargs: Dictionary with optional arguments.
+        """
+        words = (compose_word(key, value) for key, value in kwargs.items())
+        await self.protocol.writeSentence(cmd, *words)
+        response = await self.readResponse()
+        for item in response:
+            yield item
+
+    async def rawCmd(self, cmd: str, *words: str) -> AsyncResponseIter:
+        """
+        Call Api with given command and raw words.
+        End user is responsible to properly format each api word argument.
+        :param cmd: Command word. eg. /ip/address/print
+        :param args: Iterable with optional plain api arguments.
+        """
+        await self.protocol.writeSentence(cmd, *words)
+        response = await self.readResponse()
+        for item in response:
+            yield item
+
+    async def readSentence(self) -> typing.Tuple[str, ReplyDict]:
+        """
+        Read one sentence and parse words.
+        """
+        # Assuming readSentence is also an async method in the protocol
+        reply_word, words = await self.protocol.readSentence()
+        return reply_word, dict(parse_word(word) for word in words)
+
+    async def readResponse(self) -> Response:
+        """
+        Yield each sentence untill !done is received.
+
+        :throws TrapError: If one !trap is received.
+        :throws MultiTrapError: If > 1 !trap is received.
+        """
+        traps = []
+        reply_word = None
+        response = []
+        while reply_word != "!done":
+            reply_word, words = await self.readSentence()
+            if reply_word == "!trap":
+                traps.append(TrapError(**words))
+            elif reply_word in ("!re", "!done") and words:
+                response.append(words)
+
+        if len(traps) > 1:
+            raise MultiTrapError(*traps)
+        if len(traps) == 1:
+            raise traps[0]
+        return response
+
+    async def close(self) -> None:
+        await self.protocol.close()
+
+    def path(self, *path: str):
+        return AyncPath(
+            path="",
+            api=self,
+        ).join(*path)
+
+
+class AyncPath:
+    """Represents absolute command path."""
+
+    def __init__(self, path: str, api: AsyncApi):
+        self.path = path
+        self.api = api
+
+    def select(self, *keys: query.Key) -> query.AsyncQuery:
+        return query.AsyncQuery(path=self, keys=keys, api=self.api)
+
+    def __str__(self) -> str:
+        return self.path
+
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__} {self}>"
+
+    async def __aiter__(self) -> AsyncResponseIter:
+        async for response in self("print"):
+            yield response
+
+    async def __call__(self, cmd: str, **kwargs: typing.Any) -> AsyncResponseIter:
+        async for response in self.api(
+            self.join(cmd).path,
+            **kwargs,
+        ):
+            yield response
+
+    def join(self, *path: str):
+        """Join current path with one or more path strings."""
+        return AyncPath(
+            api=self.api,
+            path=pjoin("/", self.path, *path).rstrip("/"),
+        )
+
+    async def remove(self, *ids: str) -> None:
+        joined = ",".join(ids)
+        [
+            response
+            async for response in self(
+                "remove",
+                **{".id": joined},
+            )
+        ]
+
+    async def add(self, **kwargs: typing.Any) -> str:
+        response = [
+            response
+            async for response in self(
+                "add",
+                **kwargs,
+            )
+        ]
+        return response[0]["ret"]
+
+    async def update(self, **kwargs: typing.Any) -> None:
+        [
+            response
+            async for response in self(
+                "set",
+                **kwargs,
+            )
+        ]

librouteros/connections.py

@@ -2,6 +2,8 @@
 from socket import socket
 from librouteros.exceptions import ConnectionClosed
 
+from asyncio import StreamReader, StreamWriter
+
 
 class SocketTransport:
     def __init__(self, sock: socket):
@@ -28,3 +30,35 @@ def read(self, length: int) -> bytes:
 
     def close(self) -> None:
         self.sock.close()
+
+
+class AsyncSocketTransport:
+    def __init__(self, reader: StreamReader, writer: StreamWriter):
+        self.reader = reader
+        self.writer = writer
+
+    async def write(self, data: bytes) -> None:
+        """
+        Write given bytes to socket. Loop as long as every byte in
+        string is written unless exception is raised.
+        """
+        self.writer.write(data)
+        await self.writer.drain()
+
+    async def read(self, length: int) -> bytes:
+        """
+        Read as many bytes from socket as specified in length.
+        Loop as long as every byte is read unless exception is raised.
+        """
+        data = bytearray()
+        while len(data) != length:
+            data += await self.reader.read((length - len(data)))
+            if not data:
+                raise ConnectionClosed("Connection unexpectedly closed.")
+
+        # print data for debugging
+        return data
+
+    async def close(self) -> None:
+        self.writer.close()
+        await self.writer.wait_closed()

librouteros/login.py

@@ -27,3 +27,15 @@ def token(api: typing.Any, username: str, password: str) -> None:
 def plain(api: typing.Any, username: str, password: str) -> None:
     """Login using post routeros 6.43 authorization method."""
     tuple(api("/login", **{"name": username, "password": password}))
+
+
+async def async_plain(api, username, password):
+    [response async for response in api("/login", **{"name": username, "password": password})]
+
+
+async def async_token(api: typing.Any, username: str, password: str) -> None:
+    """Login using pre routeros 6.43 authorization method."""
+    sentence = [response async for response in api("/login")]
+    tok = sentence[0]["ret"]
+    encoded = encode_password(tok, password)
+    [response async for response in api("/login", **{"name": username, "response": encoded})]