BotZ/api/async_bot.py

# -*- encoding: utf-8 -*-

"""
Asyncio wrapper for AsyncOperator methods.
"""

import asyncio
from contextlib import contextmanager
import typing as T

from bot_z.async_operator import AsyncOperator
from bot_z.exceptions import OperationFailed


@contextmanager
def this_loop(loop):
    _loop = asyncio.get_event_loop()
    asyncio.set_event_loop(loop)
    yield
    asyncio.set_event_loop(_loop)


def init(base_url: T.Text, name: T.Text) -> AsyncOperator:
    """
    Initialize the stateful object.
    """
    return AsyncOperator(base_url, name)


async def login(op: AsyncOperator, username: T.Text, password: T.Text) -> bool:
    """
    Executes the login asynchronously and returns
    the AsyncOperator object.
    """
    if not op.logged_in:
        try:
            await op.login(username, password)
        except OperationFailed:
            pass
    return op.logged_in


async def logout(op: AsyncOperator) -> bool:
    """
    Executes the logout asynchronously and returns
    a bool showing the success or failure of the operation.
    """
    if op.logged_in:
        try:
            await op.logout()
        except OperationFailed:
            pass
    return op.logged_in


async def checkin(op: AsyncOperator) -> bool:
    """
    Executes the check in asynchronously and returns
    a bool showing the success or failure of the operation.
    """
    if op.logged_in and not op.checked_in:
        try:
            await op.checkin()
        except OperationFailed:
            pass
    return op.checked_in


async def checkout(op: AsyncOperator) -> bool:
    """
    Executes the checkout asynchronously and returns
    a bool showing the success or failure of the operation.
    """
    if op.logged_in and op.checked_in:
        try:
            await op.checkout()
        except OperationFailed:
            pass
    return op.checked_in


async def status(op: AsyncOperator) -> T.List[T.Optional[T.Tuple[T.Text, T.Text]]]:
    """
    Asks the list of movements asynchronously and returns a list
    of tuples containing the succession of movements. If the operation
    fails, the list is empty.
    """
    movements = []
    if op.logged_in:
        res = await op.get_movements()
        movements.extend(res)
    return movements