Skip to content

Commit

Permalink
refactor: return typed dicts in client functions
Browse files Browse the repository at this point in the history
  • Loading branch information
gmargaritis committed Feb 9, 2024
1 parent 7c4e90c commit cf3506e
Show file tree
Hide file tree
Showing 2 changed files with 94 additions and 13 deletions.
91 changes: 79 additions & 12 deletions ergani/client.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
from datetime import datetime
from typing import Any, Dict, List, Optional

import requests
Expand All @@ -10,6 +11,7 @@
CompanyOvertime,
CompanyWeeklySchedule,
CompanyWorkCard,
SubmissionResponse,
)
from ergani.utils import extract_error_message

Expand Down Expand Up @@ -37,6 +39,21 @@ def __init__(
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
"""

url = f"{self.base_url}/{endpoint}"
auth = ErganiAuthentication(self.username, self.password, self.base_url)

Expand All @@ -52,6 +69,21 @@ def _request(
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)
Expand All @@ -66,17 +98,52 @@ def _handle_response(
error_message = extract_error_message(response)
raise APIError(message=error_message, response=response, payload=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

def submit_work_card(
self, company_work_cards: List[CompanyWorkCard]
) -> Optional[Response]:
) -> 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:
An optional Response object from the Ergani API
List[SubmissionResponse]: A list of SumbmissionResponse that were parsed from the API response
Raises:
APIError: An error occurred while communicating with the Ergani API
Expand All @@ -95,19 +162,19 @@ def submit_work_card(

response = self._request("POST", endpoint, request_payload)

return response
return self._extract_submission_result(response)

def submit_overtime(
self, company_overtimes: List[CompanyOvertime]
) -> Optional[Response]:
) -> 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:
An optional Response object from the Ergani API
List[SubmissionResponse]: A list of SumbmissionResponse that were parsed from the API response
Raises:
APIError: An error occurred while communicating with the Ergani API
Expand All @@ -127,19 +194,19 @@ def submit_overtime(

response = self._request("POST", endpoint, request_payload)

return response
return self._extract_submission_result(response)

def submit_daily_schedule(
self, company_daily_schedules: List[CompanyDailySchedule]
) -> Optional[Response]:
) -> 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:
An optional Response object from the Ergani API
List[SubmissionResponse]: A list of SumbmissionResponse that were parsed from the API response
Raises:
APIError: An error occurred while communicating with the Ergani API
Expand All @@ -156,19 +223,19 @@ def submit_daily_schedule(

response = self._request("POST", endpoint, request_payload)

return response
return self._extract_submission_result(response)

def submit_weekly_schedule(
self, company_weekly_schedules: List[CompanyWeeklySchedule]
) -> Optional[Response]:
) -> 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:
An optional Response object from the Ergani API
List[SubmissionResponse]: A list of SumbmissionResponse that were parsed from the API response
Raises:
APIError: An error occurred while communicating with the Ergani API
Expand All @@ -185,4 +252,4 @@ def submit_weekly_schedule(

response = self._request("POST", endpoint, request_payload)

return response
return self._extract_submission_result(response)
16 changes: 15 additions & 1 deletion ergani/models.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from dataclasses import dataclass, field
from datetime import date, datetime, time
from typing import List, Literal, Optional
from typing import List, Literal, Optional, TypedDict

from ergani.typings import (
LateDeclarationJustificationType,
Expand Down Expand Up @@ -372,3 +372,17 @@ def serialize(self):
]
},
}

class SubmissionResponse(TypedDict):
"""
Represents a submission response from the Ergani API
Attributes:
submission_id (str): The unique identifier of the submission
protocol (str): The protocol associated with the submission
submission_date (datetime): The datetime of the submission
"""

submission_id: str
protocol: str
submission_date: datetime

0 comments on commit cf3506e

Please sign in to comment.