khushalps/cosec_automation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20260219) Updarted APIs to now show last-in time and loc also.

Browse Source
This commit is contained in:
Khushal P Soonderji
2026-02-19 14:48:41 +05:30
parent 81021a4b38
commit d8aaf942b2
7 changed files with 360 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tcaoff/async_tcaoff.py
+319 -204
View File
@@ -136,7 +136,7 @@ class AsyncTheCAOffice:
pool = 60.0,
connect = 5.0,
write = 15.0,
read = 180.0
read = 60.0
),
headers = None
)
@@ -265,170 +265,253 @@ class AsyncTheCAOffice:
# ┣┫┏┓┏┓┏┓┏┣┓┏┓┏
# ┻┛┛ ┗┻┛┗┗┛┗┗ ┛
async def branch_list(self) -> List[Dict[str, Any]] | None:
async def branch_list(
self,
raise_exception: bool = False,
) -> List[Dict[str, Any]] | None:
"""
List the existing branches.
:param raise_exception: Whether to raise any exceptions, or to suppress them.
:return: A list of dictionaries where each dictionary describes on branch. None if the API call fails.
"""
# Make the API call:
response = await self._http_client.post(
url = self.BRANCH_LIST_URL,
headers = {"X-Session-Token": self.__session_token},
json = {"idUser": self.__user_id}
)
# Start by assuming failure:
branches = None
# If the call succeeded:
if response.status_code in [200]:
response_json = response.json()
branches = response_json["data"]["rs0"]
self._printer("Branches listed.")
return branches
try:
# If the call failed:
else:
self._err_printer("Branch-list failed.")
return None
# Make the API call:
response = await self._http_client.post(
url = self.BRANCH_LIST_URL,
headers = {"X-Session-Token": self.__session_token},
json = {"idUser": self.__user_id}
)
# If the call succeeded:
if response.status_code in [200]:
response_json = response.json()
branches = response_json["data"]["rs0"]
self._printer("Branches listed.")
# If the call failed:
else:
self._err_printer("Branch-list failed.")
branches = None
# If something goes wrong:
except Exception as exception:
self._err_printer("Branch-list failed.", exception)
if raise_exception: raise
branches = None
# Done here:
return branches
async def branch_add(
self,
branch_name: str
branch_name: str,
raise_exception: bool = False,
) -> bool:
"""
Add a new branch.
:param branch_name: The name of the branch to add.
:param raise_exception: Whether to raise any exceptions, or to suppress them.
:return: True if logged out, else False.
"""
# Make the API call:
response = await self._http_client.post(
url = self.BRANCH_ADD_URL,
headers = {"X-Session-Token": self.__session_token},
json = {
"idUser": self.__user_id,
"branchName": branch_name
}
)
# Start by assuming failure:
success = False
# If the call succeeded:
if response.status_code in [200]:
self._printer(
"Branch added.",
branch_name
)
return True
try:
# If the call failed:
else:
response_json = response.json()
self._err_printer(
"Branch NOT added.",
branch_name,
response_json
# Make the API call:
response = await self._http_client.post(
url = self.BRANCH_ADD_URL,
headers = {"X-Session-Token": self.__session_token},
json = {
"idUser": self.__user_id,
"branchName": branch_name
}
)
return False
# If the call succeeded:
if response.status_code in [200]:
self._printer(
"Branch added.",
branch_name
)
success = True
# If the call failed:
else:
response_json = response.json()
self._err_printer(
"Branch NOT added.",
branch_name,
response_json
)
success = False
# If something goes wrong:
except Exception as exception:
self._err_printer("Branch NOT added.", exception, branch_name)
if raise_exception: raise
success = False
# Done here:
return success
# ┳┓
# ┃┃┏┓┏┓┏┓┏┓╋┏┳┓┏┓┏┓╋
# ┻┛┗ ┣┛┗┻┛ ┗┛┗┗┗ ┛┗┗
# ┛
async def department_list(self) -> List[Dict[str, Any]] | None:
async def department_list(
self,
raise_exception: bool = False,
) -> List[Dict[str, Any]] | None:
"""
List the existing departments.
:param raise_exception: Whether to raise any exceptions, or to suppress them.
:return: A list of dictionaries where each dictionary describes one department. None if the API call fails.
"""
# Make the API call:
response = await self._http_client.post(
url = self.DEPT_LIST_URL,
headers = {"X-Session-Token": self.__session_token},
json = {"idUser": self.__user_id}
)
# Start by assuming failure:
depts = None
# If the call succeeded:
if response.status_code in [200]:
response_json = response.json()
depts = response_json["data"]["rs0"]
self._printer("Depts. listed.")
return depts
try:
# If the call failed:
else:
self._err_printer("Dept.-list failed.")
return None
# Make the API call:
response = await self._http_client.post(
url = self.DEPT_LIST_URL,
headers = {"X-Session-Token": self.__session_token},
json = {"idUser": self.__user_id}
)
# If the call succeeded:
if response.status_code in [200]:
response_json = response.json()
depts = response_json["data"]["rs0"]
self._printer("Depts. listed.")
# If the call failed:
else:
self._err_printer("Dept.-list failed.")
depts = None
# If something goes wrong:
except Exception as exception:
self._err_printer("Dept.-list failed.", exception)
if raise_exception: raise
depts = None
# Done here:
return depts
async def department_add(
self,
department_name: str
department_name: str,
raise_exception: bool = False,
) -> bool:
"""
Add a new department.
:param department_name: The name of the department to add.
:param raise_exception: Whether to raise any exceptions, or to suppress them.
:return: True if logged out, else False.
"""
# Make the API call:
response = await self._http_client.post(
url = self.DEPT_ADD_URL,
headers = {"X-Session-Token": self.__session_token},
json = {
"idUser": self.__user_id,
"departmentName": department_name
}
)
# Start by assuming failure:
success = False
# If the call succeeded:
if response.status_code in [200]:
self._printer(
"Dept. added.",
department_name
)
return True
try:
# If the call failed:
else:
response_json = response.json()
self._err_printer(
"Dept. NOT added.",
department_name,
response_json
# Make the API call:
response = await self._http_client.post(
url = self.DEPT_ADD_URL,
headers = {"X-Session-Token": self.__session_token},
json = {
"idUser": self.__user_id,
"departmentName": department_name
}
)
return False
# If the call succeeded:
if response.status_code in [200]:
self._printer(
"Dept. added.",
department_name
)
success = True
# If the call failed:
else:
response_json = response.json()
self._err_printer(
"Dept. NOT added.",
department_name,
response_json
)
success = False
# If something goes wrong:
except Exception as exception:
self._err_printer("Dept. NOT added.", exception, department_name)
if raise_exception: raise
success = False
# Done here:
return success
# ┏┳┓
# ┃ ┏┓┏┓┏┳┓
# ┻ ┗ ┗┻┛┗┗
async def team_list(self) -> List[Dict[str, Any]] | None:
async def team_list(
self,
raise_exception: bool = False,
) -> List[Dict[str, Any]] | None:
"""
List the existing team members.
:param raise_exception: Whether to raise any exceptions, or to suppress them.
:return: A list of dictionaries where each dictionary describes one team-member. None if the API call fails.
"""
# Make the API call:
response = await self._http_client.post(
url = self.TEAM_LIST_URL,
headers = {"X-Session-Token": self.__session_token},
json = {"idUser": self.__user_id}
)
# Start by assuming failure:
teams = None
# If the call succeeded:
if response.status_code in [200]:
response_json = response.json()
teams = response_json["data"]["rs0"]
self._printer("Team listed.")
return teams
try:
# If the call failed:
else:
self._err_printer("Team-list failed.")
return None
# Make the API call:
response = await self._http_client.post(
url = self.TEAM_LIST_URL,
headers = {"X-Session-Token": self.__session_token},
json = {"idUser": self.__user_id}
)
# If the call succeeded:
if response.status_code in [200]:
response_json = response.json()
teams = response_json["data"]["rs0"]
self._printer("Team listed.")
# If the call failed:
else:
self._err_printer("Team-list failed.")
teams = None
# If something goes wrong:
except Exception as exception:
self._err_printer("Team-list filed.", exception)
if raise_exception: raise
teams = None
# Done here:
return teams
async def team_add(
self,
@@ -441,7 +524,8 @@ class AsyncTheCAOffice:
role: str,
username: str,
password: str,
applicant_notes: dict | list | str = None
applicant_notes: dict | list | str = None,
raise_exception: bool = False,
) -> bool:
"""
@@ -456,47 +540,62 @@ class AsyncTheCAOffice:
:param username: The unique username of the team member. Cannot be the same as anyone else.
:param password: The password for this team member's login.
:param applicant_notes: Optional notes about the team member.
:param raise_exception: Whether to raise any exceptions, or to suppress them.
:return: True if successful, else False.
"""
# Make the API call:
response = await self._http_client.post(
url = self.TEAM_ADD_URL,
headers = {"X-Session-Token": self.__session_token},
json = {
"branchId": branch_id,
"idDepartment": dept_id,
"reportingTo": reporting_to,
"name": team_name,
"email": email,
"phoneNo": phone_no,
"role": role,
"username": username,
"password": password,
"hierarchy": 1,
"applicantNotes": applicant_notes
}
)
# Start by assuming failure:
success = False
# If the call succeeded:
if response.status_code in [200]:
self._printer(
"Team added.",
team_name,
username,
)
return True
try:
# If the call failed:
else:
response_json = response.json()
self._err_printer(
"Team NOT added.",
team_name,
username,
response_json
# Make the API call:
response = await self._http_client.post(
url = self.TEAM_ADD_URL,
headers = {"X-Session-Token": self.__session_token},
json = {
"branchId": branch_id,
"idDepartment": dept_id,
"reportingTo": reporting_to,
"name": team_name,
"email": email,
"phoneNo": phone_no,
"role": role,
"username": username,
"password": password,
"hierarchy": 1,
"applicantNotes": applicant_notes
}
)
return False
# If the call succeeded:
if response.status_code in [200]:
self._printer(
"Team added.",
team_name,
username,
)
success = True
# If the call failed:
else:
response_json = response.json()
self._err_printer(
"Team NOT added.",
team_name,
username,
response_json
)
success = False
# If something goes wrong:
except Exception as exception:
self._err_printer("Team NOT added.", exception)
if raise_exception: raise
success = False
# Done here:
return success
# ┏┓ ┓
# ┣┫╋╋┏┓┏┓┏┫┏┓┏┓┏┏┓
@@ -508,7 +607,8 @@ class AsyncTheCAOffice:
status: Literal["P", "H", "A", "OT"],
over_time: int | float,
attendance_date: datetime.datetime | None = None,
json_notes: dict = None
json_notes: dict = None,
raise_exception: bool = False
) -> bool:
"""
@@ -522,99 +622,114 @@ class AsyncTheCAOffice:
:param over_time: The amount of over-time work in hours.
:param attendance_date: The date of the attendance. If not given, today's date will be used.
:param json_notes: Optional notes about the attendance.
:param raise_exception: Whether to raise any exceptions, or to suppress them.
:return: True if the attendance was marked, else False.
"""
# Prepare the payload:
json_payload = {
"date": (
date_time.parse_date_time(attendance_date) or
date_time.get_current_date_time()
).strftime("%Y-%m-%d"),
"idUser": user_id,
"status": status if over_time <= 0.0 else "OT",
"ot": over_time,
}
# Start by assuming failure:
success = False
# Add JSON notes if needed:
if json_notes:
json_payload["jsonNotes"] = json_notes
try:
# Make the API call:
response = await self._http_client.post(
url = self.ATTENDANCE_MARK_URL,
headers = {"X-Session-Token": self.__session_token},
json = json_payload
)
# Prepare the payload:
json_payload = {
"date": (
date_time.parse_date_time(attendance_date) or
date_time.get_current_date_time()
).strftime("%Y-%m-%d"),
"idUser": user_id,
"status": status if over_time <= 0.0 else "OT",
"ot": over_time,
}
# Debugging:
if response.status_code not in [200]:
try:
response_json = response.json()
self._printer("TCAOFF Attendance-Mark", response_json)
except Exception as e:
self._printer("TCAOFF Attendance-Mark", e, response.content)
print("Payload:", json.to_string(json_payload))
# Add JSON notes if needed:
if json_notes:
json_payload["jsonNotes"] = json_notes
# If the call succeeded:
if response.status_code in [200]:
self._printer(
"Attendance marked.",
user_id,
# Make the API call:
response = await self._http_client.post(
url = self.ATTENDANCE_MARK_URL,
headers = {"X-Session-Token": self.__session_token},
json = json_payload,
timeout = httpx.Timeout(
pool = 60.0,
connect = 5.0,
write = 15.0,
read = 10.0
)
)
return True
# If the call failed:
else:
try: response_json = response.json()
except Exception as e: response_json = response.content
self._err_printer(
"Attendance NOT added.",
user_id,
response_json
)
return False
# If the call succeeded:
if response.status_code in [200]:
self._printer("Attendance marked.", user_id, status, json_notes)
success = True
# If the call failed:
else:
try: response_json = response.json()
except Exception as e: response_json = response.content
self._err_printer("Attendance NOT added.", user_id, response_json)
success = False
# If something goes wrong:
except Exception as exception:
self._err_printer("Attendance NOT added.", exception, user_id)
if raise_exception: raise
success = False
# Done here:
return success
async def attendance_register(
self,
target_month: datetime.datetime,
raise_exception: bool = False,
) -> List[dict] | None:
"""
To get the whole attendance register for a month for all the employees of an entity.
:param target_month: The datetime which indicates the year and month in which the attendance needs to be
checked.
:param raise_exception: Whether to raise any exceptions, or to suppress them.
:return: The attendance register records if successful, else None.
"""
# Start by assuming failure:
attendance_register = None
# Make the API call:
json_payload = {"month": target_month.strftime("%Y-%m-%d")}
response = await self._http_client.post(
url = self.ATTENDANCE_REGISTER_URL,
headers = {"X-Session-Token": self.__session_token},
json = json_payload,
)
try:
# If the call succeeded:
if response.status_code in [200]:
self._printer(
"Attendance register fetched.",
target_month,
# Make the API call:
json_payload = {"month": target_month.strftime("%Y-%m-%d")}
response = await self._http_client.post(
url = self.ATTENDANCE_REGISTER_URL,
headers = {"X-Session-Token": self.__session_token},
json = json_payload,
)
attendance_register = response.json()["data"]["rs0"]
# If the call failed:
else:
try: response_data = response.json()
except Exception as e: response_data = response.content
self._err_printer(
"Attendance register NOT fetched.",
target_month,
response_data
)
# If the call succeeded:
if response.status_code in [200]:
self._printer(
"Attendance register fetched.",
target_month,
)
attendance_register = response.json()["data"]["rs0"]
# If the call failed:
else:
try: response_data = response.json()
except Exception as e: response_data = response.content
self._err_printer(
"Attendance register NOT fetched.",
target_month,
response_data
)
attendance_register = None
# If something goes wrong:
except Exception as exception:
self._err_printer(exception)
if raise_exception: raise
attendance_register = None
# Done here: