from datetime import datetime
from typing import Any, Dict, List, Optional
import requests
from requests.models import Response
from ergani.auth import ErganiAuthentication
from ergani.exceptions import APIError, AuthenticationError
from ergani.models import (
BusinessBranch,
CompanyDailySchedule,
CompanyOvertime,
CompanyWeeklySchedule,
CompanyWorkCard,
EmployerDetails,
SubmissionResponse,
)
from ergani.utils import extract_error_message, normalize_base_url
[docs]
class ErganiClient:
"""
A client for interacting with the Ergani API
Args:
username str: The username for authentication with Ergani
password str: The password for authentication with Ergani
base_url str: The base URL of the Ergani API. Defaults to "https://trialeservices.yeka.gr/WebServicesAPI/api".
"""
def __init__(
self,
username: str,
password: str,
base_url: Optional[str] = "https://trialeservices.yeka.gr/WebServicesAPI/api",
) -> None:
self.username = username
self.password = password
self.base_url = normalize_base_url(base_url)
def _request(
self, method: str, endpoint: str, payload: Optional[Dict[str, Any]] = None
) -> Optional[Response]:
"""
Sends a request to the specified endpoint using the given HTTP method and payload
Args:
method (str): The HTTP method to use for the request (e.g., 'GET', 'POST')
endpoint (str): The API endpoint to which the request should be sent to
payload (Optional[Dict[str, Any]]): The JSON-serializable dictionary to be sent as the request payload
Returns:
Optional[Response]: The response object from the requests library. Returns None for 204 No Content responses.
Raises:
Requests exceptions may be raised for network-related errors
"""
normalized_endpoint = endpoint.lstrip("/")
url = f"{self.base_url}/{normalized_endpoint}"
auth = ErganiAuthentication(self.username, self.password, self.base_url)
response = requests.request(
method,
url,
json=payload,
auth=auth,
)
return self._handle_response(response, payload)
def _handle_response(
self, response: Response, payload: Optional[Dict[str, Any]] = None
) -> Optional[Response]:
"""
Handles the HTTP response, raising exceptions for error status codes and returning the response for successful ones
Args:
response (Response): The response object to handle
payload (Optional[Dict[str, Any]]): The original request payload for inclusion in exceptions if needed
Returns:
Optional[Response]: The original response object for successful requests or None for 204 No Content responses
Raises:
APIError: An error occurred while communicating with the Ergani API
AuthenticationError: Raised if there is an authentication error with the Ergani API
"""
if response.status_code == 401:
error_message = extract_error_message(response)
raise AuthenticationError(message=error_message, response=response)
if response.status_code == 204:
return None
try:
response.raise_for_status()
return response
except requests.HTTPError:
error_message = extract_error_message(response)
raise APIError(message=error_message, response=response, payload=payload)
def _execute_service(
self, service_code: str, parameters: Optional[Dict[str, Any]] = None
) -> Optional[Response]:
request_payload = {
"ServiceCode": service_code,
"Parameters": [
{"ParameterName": name, "ParameterValue": value}
for name, value in (parameters or {}).items()
],
}
return self._request("POST", "/WebServices/ExecuteService", request_payload)
def _extract_submission_result(
self, response: Optional[Response]
) -> List[SubmissionResponse]:
"""
Extracts the submission result from the Ergani API response
Args:
response (Response): The response object from the Ergani API
Returns:
List[SubmissionResponse]: A list of submission responses parsed from the API response
Raises:
ValueError: If the response cannot be parsed into submission responses, indicating an unexpected format
"""
if not response:
return []
data = response.json()
submissions = []
for submission in data:
submission_date_str = submission["submitDate"]
submission_date = datetime.strptime(submission_date_str, "%d/%m/%Y %H:%M")
submission_response = SubmissionResponse(
submission_id=submission["id"],
protocol=submission["protocol"],
sumbmission_date=submission_date,
)
submissions.append(submission_response)
return submissions
[docs]
def submit_work_card(
self, company_work_cards: List[CompanyWorkCard]
) -> List[SubmissionResponse]:
"""
Submits work card records (check-in, check-out) for employees to the Ergani API
Args:
company_work_cards List[CompanyWorkCard]: A list of CompanyWorkCard instances to be submitted
Returns:
List[SubmissionResponse]: A list of SumbmissionResponse that were parsed from the API response
Raises:
APIError: An error occurred while communicating with the Ergani API
AuthenticationError: Raised if there is an authentication error with the Ergani API
"""
endpoint = "/Documents/WRKCardSE"
request_payload = {
"Cards": {
"Card": [
company_card.serialize() for company_card in company_work_cards
]
}
}
response = self._request("POST", endpoint, request_payload)
return self._extract_submission_result(response)
[docs]
def submit_overtime(
self, company_overtimes: List[CompanyOvertime]
) -> List[SubmissionResponse]:
"""
Submits overtime records for employees to the Ergani API
Args:
company_overtimes List[CompanyOvertime]: A list of CompanyOvertime instances to be submitted
Returns:
List[SubmissionResponse]: A list of SumbmissionResponse that were parsed from the API response
Raises:
APIError: An error occurred while communicating with the Ergani API
AuthenticationError: Raised if there is an authentication error with the Ergani API
"""
endpoint = "/Documents/OvTime"
request_payload = {
"Overtimes": {
"Overtime": [
company_overtime.serialize()
for company_overtime in company_overtimes
]
}
}
response = self._request("POST", endpoint, request_payload)
return self._extract_submission_result(response)
[docs]
def submit_daily_schedule(
self, company_daily_schedules: List[CompanyDailySchedule]
) -> List[SubmissionResponse]:
"""
Submits schedule records that are updated on a daily basis for employees to the Ergani API
Args:
company_daily_schedules List[CompanyDailySchedule]: A list of CompanyDailySchedule instances to be submitted
Returns:
List[SubmissionResponse]: A list of SumbmissionResponse that were parsed from the API response
Raises:
APIError: An error occurred while communicating with the Ergani API
AuthenticationError: Raised if there is an authentication error with the Ergani API
"""
endpoint = "/Documents/WTODaily"
request_payload = {
"WTOS": {
"WTO": [schedule.serialize() for schedule in company_daily_schedules]
}
}
response = self._request("POST", endpoint, request_payload)
return self._extract_submission_result(response)
[docs]
def submit_weekly_schedule(
self, company_weekly_schedules: List[CompanyWeeklySchedule]
) -> List[SubmissionResponse]:
"""
Submits weekly schedule records for employees to the Ergani API
Args:
company_weekly_schedules List[CompanyWeeklySchedule]: A list of CompanyWeeklySchedule instances to be submitted
Returns:
List[SubmissionResponse]: A list of SumbmissionResponse that were parsed from the API response
Raises:
APIError: An error occurred while communicating with the Ergani API
AuthenticationError: Raised if there is an authentication error with the Ergani API
"""
endpoint = "/Documents/WTOWeek"
request_payload = {
"WTOS": {
"WTO": [schedule.serialize() for schedule in company_weekly_schedules]
}
}
response = self._request("POST", endpoint, request_payload)
return self._extract_submission_result(response)
[docs]
def get_services_list(self) -> Optional[Response]:
"""
Fetches the available services list from the Ergani API.
Returns:
Optional[Response]: The raw response returned by the services list endpoint.
Raises:
APIError: An error occurred while communicating with the Ergani API
AuthenticationError: Raised if there is an authentication error with the Ergani API
"""
return self._request("GET", "/WebServices/ServicesList", None)
[docs]
def get_branch_details(self) -> List[BusinessBranch]:
"""
Fetches the authenticated employer's branch details from the Ergani API.
Returns:
List[BusinessBranch]: The parsed branch detail entries.
Raises:
APIError: An error occurred while communicating with the Ergani API
AuthenticationError: Raised if there is an authentication error with the Ergani API
ValueError: The response payload could not be parsed as a branch list
"""
response = self._execute_service("EX_BASE_02")
payload = None
if response:
try:
payload = response.json()
except ValueError as error:
raise ValueError("EX_BASE_02 returned a non-JSON response") from error
return BusinessBranch.parse_many(payload)
[docs]
def get_employer_details(self) -> EmployerDetails:
"""
Fetches employer details from the Ergani API.
Returns:
EmployerDetails: Employer details parsed from the EX_BASE_01 response.
Raises:
APIError: An error occurred while communicating with the Ergani API
AuthenticationError: Raised if there is an authentication error with the Ergani API
ValueError: Raised if the query payload cannot be parsed into an employer details object
"""
response = self._execute_service("EX_BASE_01")
if response is None:
return EmployerDetails.parse({})
try:
payload = response.json()
except ValueError as error:
raise ValueError("EX_BASE_01 returned a non-JSON response") from error
return EmployerDetails.parse(payload)