"""Implementation of client for communicating with the server."""
from shv import RpcMethodNotFoundError, RpcUrl, SHVType, SimpleClient
[docs]
class ClockInClient:
"""Ellclockin client for communicating with the server."""
[docs]
async def clockin_project(self, project: str) -> SHVType:
"""Clockin the current user to the specified project.
:param project: name of the project
"""
return await self.client.call(f"clockin/projects/{project}", "work")
[docs]
async def clockout_project(self) -> SHVType:
"""Clockout the current user from their current project."""
return await self.client.call("clockin/projects", "clockout")
[docs]
async def clockout_retroactive(self, retro_date: float) -> SHVType:
"""Retroactively clockout the current user from their current project.
:param retro_date: clockout time timestamp
"""
return await self.client.call("clockin/projects", "retroactive", retro_date)
[docs]
async def connect(self, url: RpcUrl) -> None:
"""Connect to the server.
:param url: URL of the server
"""
self.client = await SimpleClient.connect(url)
assert self.client is not None
res = await self.client.call(".app", "name")
print(f"Connected to: {res!r}")
[docs]
async def create_new_project(self, project: str) -> SHVType:
"""Create a new project with the specified name.
:param project: name of the new project
"""
return await self.client.call("clockin/projects", "new", project)
[docs]
async def create_new_worker(self, worker: str) -> SHVType:
"""Create a new worker with the specified name.
:param project: name of the new worker
"""
return await self.client.call("clockin/workers", "new", worker)
[docs]
async def disconnect(self) -> None:
"""Disconnect from the server."""
await self.client.disconnect()
[docs]
async def get_all_projects(self) -> str:
"""Get the names of all projects."""
return str(await self.client.call("clockin/projects", "ls"))
[docs]
async def get_all_workers(self) -> str:
"""Get the name of all workers."""
return str(await self.client.call("clockin/workers", "ls"))
[docs]
async def get_user_project_contributions(self, project: str, user: str) -> str:
"""Get the contributions of the specified user in the specified project.
:param project: name of the specified project
:param user: username of the specified user
"""
return str(
await self.client.call(f"clockin/projects/{project}", "contributions", user)
)
[docs]
async def is_project_complete(self, project: str) -> bool:
"""Check whether the specified project is complete or not.
:param project: name of the specified project
:return: ``True`` in case the project is complete and ``False`` otherwise
"""
return bool(
await self.client.call(f"clockin/projects/{project}", "checkComplete")
)
[docs]
async def ls(self, path: str = "") -> None:
"""Print a list of all child nodes of the node on the specified path.
:param path: SHV path to the node we want children to be listed for
"""
try:
response = str(await self.client.call(path, "ls"))
except RpcMethodNotFoundError as e:
print(f"No such path found: {e}")
print(response)