93 lines
2.2 KiB
Python
93 lines
2.2 KiB
Python
# -*- 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
|