Skip to content

Introduce multiple record submissions #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Jan 30, 2024
Merged

Introduce multiple record submissions #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 7 additions & 11 deletions ergani/auth.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@

import requests
from requests.auth import AuthBase
from requests.exceptions import RequestException
from requests.models import PreparedRequest

from ergani.exceptions import AuthenticationError
Expand All @@ -29,22 +28,19 @@ def __call__(self, request: PreparedRequest) -> PreparedRequest:
request.headers["Authorization"] = f"Bearer {self.access_token}"
return request

def _authenticate(self) -> Optional[str]:
def _authenticate(self) -> str:
endpoint = "/Authentication"
payload = {
"Username": self.username,
"Password": self.password,
"UserType": "01",
}

try:
response = requests.post(f"{self.base_url}{endpoint}", json=payload)
response = requests.post(f"{self.base_url}{endpoint}", json=payload)

if response.status_code != 200:
error_message = extract_error_message(response)
raise AuthenticationError(message=error_message, response=response)
if response.status_code != 200:
error_message = extract_error_message(response)
raise AuthenticationError(message=error_message, response=response)

token = response.json().get("accessToken")
return token
except RequestException:
raise AuthenticationError()
token = response.json()["accessToken"]
return token
186 changes: 49 additions & 137 deletions ergani/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,18 @@
from typing import Any, Dict, List, Optional, Tuple
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 (
CompanyDailySchedule,
CompanyOvertime,
CompanyWeeklySchedule,
CompanyWorkCard,
)
from ergani.utils import extract_error_message

from .auth import ErganiAuthentication


class ErganiClient:
def __init__(
Expand All @@ -24,11 +29,13 @@ def _request(
self, method: str, endpoint: str, payload: Optional[Dict[str, Any]] = None
) -> Optional[Response]:
url = f"{self.base_url}/{endpoint}"
auth = ErganiAuthentication(self.username, self.password, self.base_url)

response = requests.request(
method,
url,
json=payload,
auth=ErganiAuthentication(self.username, self.password, self.base_url),
auth=auth,
)

return self._handle_response(response, payload)
Expand All @@ -51,41 +58,18 @@ def _handle_response(
raise APIError(message=error_message, response=response, payload=payload)

def submit_work_card(
self,
employer_tax_identification_number: str,
business_branch_number: str,
employee_tax_identification_number: str,
employee_last_name: str,
employee_first_name: str,
work_card_movement_type: int,
work_card_submission_date: str,
work_card_movement_datetime: str,
comments: Optional[str] = "",
late_declaration_justification_code: Optional[str] = "",
self, company_work_cards: List[CompanyWorkCard]
) -> Optional[Response]:
"""
Submits work cards records (check-in, check-out) for employees to the Ergani API
"""

endpoint = "/Documents/WRKCardSE"

request_payload = {
"Cards": {
"Card": [
{
"f_afm_ergodoti": employer_tax_identification_number,
"f_aa": business_branch_number,
"f_comments": comments,
"Details": {
"CardDetails": [
{
"f_afm": employee_tax_identification_number,
"f_eponymo": employee_last_name,
"f_onoma": employee_first_name,
"f_type": work_card_movement_type,
"f_reference_date": work_card_submission_date,
"f_date": work_card_movement_datetime,
"f_aitiologia": late_declaration_justification_code,
}
]
},
}
company_card.serialize() for company_card in company_work_cards
]
}
}
Expand All @@ -95,73 +79,19 @@ def submit_work_card(
return response

def submit_overtime(
self,
business_branch_number: str,
sepe_service_code: str,
business_primary_activity_code: str,
business_branch_activity_code: str,
kallikratis_municipal_code: str,
legal_representative_tax_identification_number: str,
employee_tax_identification_number: str,
employee_social_security_number: str,
employee_last_name: str,
employee_first_name: str,
overtime_date: str,
overtime_start_time: str,
overtime_end_time: str,
overtime_cancellation: str,
employee_profession_code: str,
overtime_justification_code: str,
weekly_workdays_number: str,
related_protocol_number: Optional[str] = "",
related_protocol_date: Optional[str] = "",
employer_organization: Optional[str] = "",
business_secondary_activity_code_1: Optional[str] = "",
business_secondary_activity_code_2: Optional[str] = "",
business_secondary_activity_code_3: Optional[str] = "",
business_secondary_activity_code_4: Optional[str] = "",
comments: Optional[str] = "",
asee_approval: Optional[str] = "",
self, company_overtimes: List[CompanyOvertime]
) -> Optional[Response]:
"""
Submits overtime records for employees to the Ergani API
"""

endpoint = "/Documents/OvTime"

request_payload = {
"Overtimes": {
"Overtime": [
{
"f_aa_pararthmatos": business_branch_number,
"f_rel_protocol": related_protocol_number,
"f_rel_date": related_protocol_date,
"f_ypiresia_sepe": sepe_service_code,
"f_ergodotikh_organwsh": employer_organization,
"f_kad_kyria": business_primary_activity_code,
"f_kad_deyt_1": business_secondary_activity_code_1,
"f_kad_deyt_2": business_secondary_activity_code_2,
"f_kad_deyt_3": business_secondary_activity_code_3,
"f_kad_deyt_4": business_secondary_activity_code_4,
"f_kad_pararthmatos": business_branch_activity_code,
"f_kallikratis_pararthmatos": kallikratis_municipal_code,
"f_comments": comments,
"f_afm_proswpoy": legal_representative_tax_identification_number,
"Ergazomenoi": {
"OvertimeErgazomenosDate": [
{
"f_afm": employee_tax_identification_number,
"f_amka": employee_social_security_number,
"f_eponymo": employee_last_name,
"f_onoma": employee_first_name,
"f_date": overtime_date,
"f_from": overtime_start_time,
"f_to": overtime_end_time,
"f_cancellation": overtime_cancellation,
"f_step": employee_profession_code,
"f_reason": overtime_justification_code,
"f_weekdates": weekly_workdays_number,
"f_asee": asee_approval,
}
]
},
}
company_overtime.serialize()
for company_overtime in company_overtimes
]
}
}
Expand All @@ -171,54 +101,36 @@ def submit_overtime(
return response

def submit_daily_schedule(
self,
business_branch_number: str,
employee_tax_identification_number: str,
employee_last_name: str,
employee_first_name: str,
schedule_date: str,
work_type: str,
workday_start_time: str,
workday_end_time: str,
related_protocol_number: Optional[str] = "",
related_protocol_date: Optional[str] = "",
comments: Optional[str] = "",
schedule_start_date: Optional[str] = "",
schedule_end_date: Optional[str] = "",
self, company_daily_schedules: List[CompanyDailySchedule]
) -> Optional[Response]:
"""
Submits schedule records that are updated on a daily basis for employees to the Ergani API
"""

endpoint = "/Documents/WTODaily"

request_payload = {
"WTOS": {
"WTO": [
{
"f_aa_pararthmatos": business_branch_number,
"f_rel_protocol": related_protocol_number,
"f_rel_date": related_protocol_date,
"f_comments": comments,
"f_from_date": schedule_start_date,
"f_to_date": schedule_end_date,
"Ergazomenoi": {
"ErgazomenoiWTO": [
{
"f_afm": employee_tax_identification_number,
"f_eponymo": employee_last_name,
"f_onoma": employee_first_name,
"f_date": schedule_date,
"ErgazomenosAnalytics": {
"ErgazomenosWTOAnalytics": [
{
"f_type": work_type,
"f_from": workday_start_time,
"f_to": workday_end_time,
}
]
},
}
]
},
}
]
"WTO": [schedule.serialize() for schedule in company_daily_schedules]
}
}

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

return response

def submit_weekly_schedule(
self, company_weekly_schedules: List[CompanyWeeklySchedule]
) -> Optional[Response]:
"""
Submits weekly schedule records for employees to the Ergani API
"""

endpoint = "/Documents/WTOWeek"

request_payload = {
"WTOS": {
"WTO": [schedule.serialize() for schedule in company_weekly_schedules]
}
}

Expand Down
15 changes: 9 additions & 6 deletions ergani/exceptions.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,13 +19,16 @@ def __init__(
self.payload = payload

def __str__(self) -> str:
if not self.message:
if self.response.status_code >= 500:
self.message = "Service unavailable, please try again later"
if self.response.status_code == 400:
self.message = "Please check your inputs and try again"
if self.response:
if not self.message:
if self.response.status_code >= 500:
self.message = "Service unavailable, please try again later"
if self.response.status_code == 400:
self.message = "Please check your inputs and try again"

return f"Status code {self.response.status_code}. Error message: {self.message}"
return f"Status code {self.response.status_code}. Error message: {self.message}"

return f"Error message: {self.message}"


class AuthenticationError(Error):
Expand Down
Loading