Add /movements api.

api/async_bot.py

@@ -77,3 +77,16 @@ async def checkout(op: AsyncOperator) -> bool:
         except OperationFailed:
             pass
     return not 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

api/rest.py

@@ -5,13 +5,14 @@ The REST endpoints.
 """
 
 from aiohttp import web
+import datetime
 import logging
 import typing as T
 
 from aiohttp_session import get_session
 
 from bot_z.async_operator import AsyncOperator
-from api.async_bot import login, logout, checkin, checkout
+from api.async_bot import login, logout, checkin, checkout, status
 from api import BASE_URI
 
 
@@ -61,3 +62,16 @@ async def logout_handler(request: web.Request) -> web.Response:
     alog.debug("logout result: %s", res)
     # FIX: assess if better to invalidate session and dump the browser session.
     return web.json_response({"logged_in": res}, status=200)
+@routes.get("/movements")
+async def movements_handle(request: web.Request) -> web.Response:
+    alog.debug("movements")
+    session = await get_session(request)
+    op = OPERATORS.get(session.get("async_operator"))
+    if not op:
+        return web.json_response({"error": "No session"}, status=404)
+    res = await status(op)
+    alog.debug("movements result: %s", res)
+    if not res:
+        return web.json_response({"error": "No movements found"}, status=404)
+    resp_data = dict((i[1], i[0]) for i in res)  # type: ignore
+    return web.json_response(resp_data, status=200)

bot_z/async_operator.py

@@ -83,6 +83,16 @@ class AsyncOperator(object):
             raise OperationFailed("Failed to checkout.")
         alog.info("Checked out [%s]", self.name)
 
+    async def get_movements(self) -> T.List[T.Optional[T.Tuple[T.Text, T.Text]]]:
+        """
+        Retrieves the list of movements as a list of tuples.
+        The list may be empty.
+        """
+        alog.debug("Retrieving the list of movements [%s]", self.name)
+        res = await _push_to_loop(self.loop, self.executor, self.op.get_movements)
+        alog.info("List of movements [%s]: %s", self.name, res)
+        return res
+
     @property
     def logged_in(self) -> bool:
         return self.op.logged_in