BotZ/api/async_bot.py

93 lines
2.2 KiB
Python
Raw Normal View History

# -*- encoding: utf-8 -*-
"""
Asyncio wrapper for AsyncOperator methods.
"""
import asyncio
from contextlib import contextmanager
2019-08-01 16:12:55 +02:00
import typing as T
from bot_z.async_operator import AsyncOperator
2019-08-01 00:28:09 +02:00
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)
2019-08-01 16:12:55 +02:00
def init(base_url: T.Text, name: T.Text) -> AsyncOperator:
"""
Initialize the stateful object.
"""
return AsyncOperator(base_url, name)
2019-08-01 16:12:55 +02:00
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:
2019-08-01 00:28:09 +02:00
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:
2019-08-01 00:28:09 +02:00
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:
2019-08-01 00:28:09 +02:00
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:
2019-08-01 00:28:09 +02:00
try:
await op.checkout()
except OperationFailed:
pass
return not op.checked_in
2019-08-01 16:17:14 +02:00
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