Skip to content

Commit

Permalink
ruff format with updated version #102
Browse files Browse the repository at this point in the history
  • Loading branch information
bensteUEM committed Oct 7, 2024
1 parent aaa315f commit 493f6e8
Show file tree
Hide file tree
Showing 16 changed files with 222 additions and 104 deletions.
4 changes: 3 additions & 1 deletion churchtools_api/calendar.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,9 @@ def get_calendar_appointments(self, calendar_ids: list, **kwargs) -> list[dict]:
if response.status_code == 200:
response_content = json.loads(response.content)
response_data = self.combine_paginated_response_data(
response_content, url=url, headers=headers,
response_content,
url=url,
headers=headers,
)

result = (
Expand Down
8 changes: 7 additions & 1 deletion churchtools_api/churchtools_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,13 @@ class ChurchToolsApi(
ChurchToolsApiResources: all functions used for resources
"""

def __init__(self, domain:str, ct_token:Optional[str]=None, ct_user:Optional[str]=None, ct_password:Optional[str]=None) -> None:
def __init__(
self,
domain: str,
ct_token: Optional[str] = None,
ct_user: Optional[str] = None,
ct_password: Optional[str] = None,
) -> None:
"""Setup of a ChurchToolsApi object for the specified ct_domain using a token login.
Arguments:
Expand Down
5 changes: 4 additions & 1 deletion churchtools_api/churchtools_api_abstract.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,10 @@ def __init__(self) -> None:
self.domain = None

def combine_paginated_response_data(
self, response_content: dict, url: str, **kwargs,
self,
response_content: dict,
url: str,
**kwargs,
) -> dict:
"""Helper function which combines data for requests in case of paginated responses.
Expand Down
46 changes: 31 additions & 15 deletions churchtools_api/events.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,9 +64,11 @@ def get_events(self, **kwargs) -> list[dict]:
params["to"] = to_
if from_ == to_:
logger.warning(
"As of 29. Sept 2024 there is a known bug which will not return results for same day searches! - reported to CT team as issue 130629 - github #109",
"As of 29. Sept 2024 there is a known bug which will not return results for same day searches! - reported to CT team as issue 130629 - github #109",
)
params["to"] = (kwargs["to_"] + timedelta(days=1)).strftime("%Y-%m-%d") #TODO #109 CT issue
params["to"] = (kwargs["to_"] + timedelta(days=1)).strftime(
"%Y-%m-%d"
) # TODO #109 CT issue
elif "to_" in kwargs:
logger.warning("Use of to_ is only allowed together with from_")
if "canceled" in kwargs:
Expand All @@ -87,7 +89,10 @@ def get_events(self, **kwargs) -> list[dict]:
if response.status_code == 200:
response_content = json.loads(response.content)
response_data = self.combine_paginated_response_data(
response_content, url=url, headers=headers, params=params,
response_content,
url=url,
headers=headers,
params=params,
)
return [response_data] if isinstance(response_data, dict) else response_data
logger.warning(
Expand All @@ -98,7 +103,9 @@ def get_events(self, **kwargs) -> list[dict]:
return None

def get_event_by_calendar_appointment(
self, appointment_id: int, start_date: str | datetime,
self,
appointment_id: int,
start_date: str | datetime,
) -> dict:
"""This method is a helper to retrieve an event for a specific calendar appointment
including it's event services.
Expand Down Expand Up @@ -136,7 +143,7 @@ def get_event_by_calendar_appointment(
)
return None

def get_AllEventData_ajax(self, eventId)->dict:
def get_AllEventData_ajax(self, eventId) -> dict:
"""Reverse engineered function from legacy AJAX API which is used to get all event data for one event.
Required to read special params not yet included in REST getEvents()
Expand Down Expand Up @@ -228,7 +235,8 @@ def set_event_services_counts_ajax(self, eventId, serviceId, servicesCount) -> b
services = self.get_services(returnAsDict=True)
serviceGroupId = services[serviceId]["serviceGroupId"]
servicesOfServiceGroup = self.get_event_services_counts_ajax(
eventId, serviceGroupId=serviceGroupId,
eventId,
serviceGroupId=serviceGroupId,
)
# set new assignment
servicesOfServiceGroup[serviceId] = servicesCount
Expand Down Expand Up @@ -266,7 +274,7 @@ def set_event_services_counts_ajax(self, eventId, serviceId, servicesCount) -> b
)
return False

def get_event_admins_ajax(self, eventId:int)->list:
def get_event_admins_ajax(self, eventId: int) -> list:
"""Get the admin id list of an event using legacy AJAX API.
Params:
Expand All @@ -284,7 +292,7 @@ def get_event_admins_ajax(self, eventId:int)->list:
logger.info("No admins found because event not found")
return []

def set_event_admins_ajax(self, eventId:int, admin_ids:list) ->bool:
def set_event_admins_ajax(self, eventId: int, admin_ids: list) -> bool:
"""Set the admin id list of an event using legacy AJAX API.
Parameters:
Expand Down Expand Up @@ -318,7 +326,7 @@ def set_event_admins_ajax(self, eventId:int, admin_ids:list) ->bool:
)
return False

def get_event_agenda(self, eventId:int)->list:
def get_event_agenda(self, eventId: int) -> list:
"""Retrieve agenda for event by ID from ChurchTools
Params:
eventId: number of the event
Expand All @@ -340,7 +348,9 @@ def get_event_agenda(self, eventId:int)->list:
)
return None

def export_event_agenda(self, target_format, target_path="./downloads", **kwargs)->bool:
def export_event_agenda(
self, target_format, target_path="./downloads", **kwargs
) -> bool:
"""Exports the agenda as zip file for imports in presenter-programs.
Parameters:
Expand Down Expand Up @@ -408,15 +418,19 @@ def export_event_agenda(self, target_format, target_path="./downloads", **kwargs
}

response = self.session.post(
url=url, params=params, headers=headers, json=json_data,
url=url,
params=params,
headers=headers,
json=json_data,
)
result_ok = False
if response.status_code == 200:
response_content = json.loads(response.content)
agenda_data = response_content["data"].copy()
logger.debug(f"Agenda package found {response_content}")
result_ok = self.file_download_from_url(
"{}/{}".format(self.domain, agenda_data["url"]), target_path,
"{}/{}".format(self.domain, agenda_data["url"]),
target_path,
)
if result_ok:
logger.debug("download finished")
Expand Down Expand Up @@ -445,7 +459,8 @@ def get_event_agenda_docx(self, agenda, **kwargs):
heading += "- Draft" if not agenda["isFinal"] else ""
document.add_heading(heading)
modifiedDate = datetime.strptime(
agenda["meta"]["modifiedDate"], "%Y-%m-%dT%H:%M:%S%z",
agenda["meta"]["modifiedDate"],
"%Y-%m-%dT%H:%M:%S%z",
)
modifiedDate2 = modifiedDate.astimezone().strftime("%a %d.%m (%H:%M:%S)")
document.add_paragraph(
Expand Down Expand Up @@ -496,7 +511,8 @@ def get_event_agenda_docx(self, agenda, **kwargs):
if (
len(item["responsible"]) > 0
and len(item["responsible"]["persons"]) == 0
and len(item["responsible"]["text"]) > 0):
and len(item["responsible"]["text"]) > 0
):
responsible_list.append(
item["responsible"]["text"]
+ " (Person statt Rolle in ChurchTools hinterlegt!)",
Expand Down Expand Up @@ -540,7 +556,7 @@ def get_persons_with_service(self, eventId: int, serviceId: int) -> list[dict]:
service for service in eventServices if service["serviceId"] == serviceId
]

def get_event_masterdata(self, **kwargs)->list | list[list] | dict | list[dict]:
def get_event_masterdata(self, **kwargs) -> list | list[list] | dict | list[dict]:
"""Function to get the Masterdata of the event module.
This information is required to map some IDs to specific items.
Expand Down
27 changes: 17 additions & 10 deletions churchtools_api/files.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,11 @@ def __init__(self) -> None:

def file_upload(
self,
source_filepath:str,
domain_type:str,
domain_identifier:int,
custom_file_name:Optional[str]=None,
overwrite:bool=False,
source_filepath: str,
domain_type: str,
domain_identifier: int,
custom_file_name: Optional[str] = None,
overwrite: bool = False,
) -> bool:
"""Helper function to upload an attachment to any module of ChurchTools.
Expand Down Expand Up @@ -95,8 +95,11 @@ def file_upload(
return False

def file_delete(
self, domain_type:str, domain_identifier:int, filename_for_selective_delete:Optional[str]=None,
)->bool:
self,
domain_type: str,
domain_identifier: int,
filename_for_selective_delete: Optional[str] = None,
) -> bool:
"""Helper function to delete ALL attachments of any specified module of ChurchTools#
or identifying individual file_name_ids and deleting specifc files only.
Expand Down Expand Up @@ -130,8 +133,12 @@ def file_delete(
return response.status_code == 204 # success code for delete action upload

def file_download(
self, filename:str, domain_type:str, domain_identifier:str, target_path:str="./downloads",
)->bool:
self,
filename: str,
domain_type: str,
domain_identifier: str,
target_path: str = "./downloads",
) -> bool:
"""Retrieves the first file from ChurchTools for specific filename, domain_type and domain_identifier from churchtools.
Params:
Expand Down Expand Up @@ -185,7 +192,7 @@ def file_download(
)
return None

def file_download_from_url(self, file_url:str, target_path:str) -> bool:
def file_download_from_url(self, file_url: str, target_path: str) -> bool:
"""Retrieves file from ChurchTools for specific file_url from churchtools.
This function is used by file_download(...).
Expand Down
48 changes: 31 additions & 17 deletions churchtools_api/groups.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,11 +45,11 @@ def get_groups(self, **kwargs) -> list[dict]:
response_content = json.loads(response.content)

response_data = self.combine_paginated_response_data(
response_content, url=url, headers=headers,
)
return (
[response_data] if isinstance(response_data, dict) else response_data
response_content,
url=url,
headers=headers,
)
return [response_data] if isinstance(response_data, dict) else response_data
logger.warning(
"%s Something went wrong fetching groups: %s",
response.status_code,
Expand Down Expand Up @@ -99,7 +99,9 @@ def get_group_statistics(self, group_id: int) -> dict:
response_content = json.loads(response.content)

response_data = self.combine_paginated_response_data(
response_content, url=url, headers=headers,
response_content,
url=url,
headers=headers,
)
logger.debug(
f"First response of Group Statistics successful {response_content}",
Expand All @@ -113,7 +115,11 @@ def get_group_statistics(self, group_id: int) -> dict:
return None

def create_group(
self, name: str, group_status_id: int, grouptype_id: int, **kwargs,
self,
name: str,
group_status_id: int,
grouptype_id: int,
**kwargs,
) -> dict:
"""Create a new group.
Expand Down Expand Up @@ -157,7 +163,9 @@ def create_group(
if response.status_code == 201:
response_content = json.loads(response.content)
response_data = self.combine_paginated_response_data(
response_content, url=url, headers=headers,
response_content,
url=url,
headers=headers,
)
logger.debug(
f"First response of Create Group successful {response_content}",
Expand Down Expand Up @@ -309,11 +317,11 @@ def get_group_members(self, group_id: int, **kwargs) -> list[dict]:
response_content = json.loads(response.content)

response_data = self.combine_paginated_response_data(
response_content, url=url, headers=headers,
)
return (
[response_data] if isinstance(response_data, dict) else response_data
response_content,
url=url,
headers=headers,
)
return [response_data] if isinstance(response_data, dict) else response_data

logger.warning(
"%s Something went wrong fetching group members: %s",
Expand All @@ -323,7 +331,10 @@ def get_group_members(self, group_id: int, **kwargs) -> list[dict]:
return None

def get_groups_members(
self, group_ids: Optional[list[int]] = None, with_deleted: bool = False, **kwargs,
self,
group_ids: Optional[list[int]] = None,
with_deleted: bool = False,
**kwargs,
) -> list[dict]:
"""Access to /groups/members to lookup group memberships
Similar to get_group_members but not specific to a single group.
Expand Down Expand Up @@ -353,7 +364,10 @@ def get_groups_members(
response_content = json.loads(response.content)

response_data = self.combine_paginated_response_data(
response_content, url=url, headers=headers, params=params,
response_content,
url=url,
headers=headers,
params=params,
)
result_list = (
[response_data] if isinstance(response_data, dict) else response_data
Expand Down Expand Up @@ -460,11 +474,11 @@ def get_group_roles(self, group_id: int) -> list[dict]:
response_content = json.loads(response.content)

response_data = self.combine_paginated_response_data(
response_content, url=url, headers=headers,
)
return (
[response_data] if isinstance(response_data, dict) else response_data
response_content,
url=url,
headers=headers,
)
return [response_data] if isinstance(response_data, dict) else response_data
logger.warning(
"%s Something went wrong fetching group roles: %s",
response.status_code,
Expand Down
10 changes: 8 additions & 2 deletions churchtools_api/persons.py
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,10 @@ def get_persons(self, **kwargs) -> list[dict]:
)

response_data = self.combine_paginated_response_data(
response_content, url=url, headers=headers, params=params,
response_content,
url=url,
headers=headers,
params=params,
)
response_data = (
[response_data] if isinstance(response_data, dict) else response_data
Expand All @@ -75,7 +78,10 @@ def get_persons(self, **kwargs) -> list[dict]:
return None

def get_persons_masterdata(
self, resultClass: Optional[str] = None, returnAsDict: bool = False, **kwargs,
self,
resultClass: Optional[str] = None,
returnAsDict: bool = False,
**kwargs,
) -> dict[list[dict]]:
"""Function to get the Masterdata of the persons module
This information is required to map some IDs to specific items.
Expand Down
Loading

0 comments on commit 493f6e8

Please sign in to comment.