Skip to content

Commit

Permalink
Move logic to helper functions to comply with cyclomatic complexity t…
Browse files Browse the repository at this point in the history
…hreshold.
  • Loading branch information
FriedrichtenHagen committed Nov 28, 2024
1 parent 343b34a commit 0b769ca
Showing 1 changed file with 39 additions and 18 deletions.
57 changes: 39 additions & 18 deletions src/personio_py/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -186,30 +186,51 @@ def request_paginated(self, path: str, method='GET', params: Dict[str, Any] = No
response = self.request_json(path, method, params, data, auth_rotation=auth_rotation)
resp_data = response.get('data')
if resp_data:
if url_type == 'absence':
data_acc.extend(resp_data)
if response['metadata']['current_page'] == response['metadata']['total_pages']:
break
else:
params['offset'] += 1
elif url_type == 'attendance':
if params['offset'] >= response['metadata']['total_elements']:
break
else:
data_acc.extend(resp_data)
params['offset'] += limit
elif url_type == 'employees':
data_acc.extend(resp_data)
if response['metadata']['current_page'] + 1 == response['metadata']['total_pages']:
break
else:
params['offset'] += 1
if url_type == 'absence' and self._handle_absence_pagination(
data_acc, resp_data, response, params
):
break
elif url_type == 'attendance' and self._handle_attendance_pagination(
data_acc, resp_data, response, params
):
break
elif url_type == 'employees' and self._handle_employees_pagination(
data_acc, resp_data, response, params
):
break
else:
break
# return the accumulated data
response['data'] = data_acc
return response

def _handle_absence_pagination(self, data_acc, resp_data, response, params) -> bool:
"""Handle pagination logic for absences."""
data_acc.extend(resp_data)
if response['metadata']['current_page'] == response['metadata']['total_pages']:
return True
params['offset'] += 1
return False

def _handle_attendance_pagination(self, data_acc, resp_data, response, params) -> bool:
"""Handle pagination logic for attendance."""
if params['offset'] >= response['metadata']['total_elements']:
return True
data_acc.extend(resp_data)
params['offset'] += params['limit']
return False

def _handle_employees_pagination(self, data_acc, resp_data, response, params):
"""Handle pagination logic for employees."""
data_acc.extend(resp_data)
total_pages = response['metadata']['total_pages']
current_page = response['metadata']['current_page']
if current_page + 1 == total_pages:
return True
else:
params['offset'] += 1
return False

def request_image(self, path: str, method='GET', params: Dict[str, Any] = None,
auth_rotation=False) -> Optional[bytes]:
"""
Expand Down

0 comments on commit 0b769ca

Please sign in to comment.