admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(calendar): Remove try/except in calendar API

Browse Source
This commit is contained in:
Chris Coutinho
2025-07-30 11:03:01 +02:00
parent 2a5b12343c
commit b81fe6dfa0
1 changed files with 106 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/client/calendar.py
+106 -157
View File
@@ -46,99 +46,75 @@ class CalendarClient(BaseNextcloudClient):
"Accept": "application/xml", "Accept": "application/xml",
} }
try: response = await self._client.request(
response = await self._client.request( "PROPFIND", caldav_path, content=propfind_body, headers=headers
"PROPFIND", caldav_path, content=propfind_body, headers=headers )
response.raise_for_status()
# Parse XML response
root = ET.fromstring(response.content)
calendars = []
for response_elem in root.findall(".//{DAV:}response"):
href = response_elem.find(".//{DAV:}href")
if href is None:
continue
href_text = href.text or ""
if not href_text.endswith("/"):
continue # Skip non-calendar resources
# Extract calendar name from href
calendar_name = href_text.rstrip("/").split("/")[-1]
if not calendar_name or calendar_name == self.username:
continue
# Get properties
propstat = response_elem.find(".//{DAV:}propstat")
if propstat is None:
continue
prop = propstat.find(".//{DAV:}prop")
if prop is None:
continue
# Check if it's a calendar resource
resourcetype = prop.find(".//{DAV:}resourcetype")
is_calendar = (
resourcetype is not None
and resourcetype.find(".//{urn:ietf:params:xml:ns:caldav}calendar")
is not None
) )
response.raise_for_status()
# Parse XML response if not is_calendar:
root = ET.fromstring(response.content) continue
calendars = []
for response_elem in root.findall(".//{DAV:}response"): # Extract calendar properties
href = response_elem.find(".//{DAV:}href") displayname_elem = prop.find(".//{DAV:}displayname")
if href is None: displayname = (
continue displayname_elem.text if displayname_elem is not None else calendar_name
)
href_text = href.text or "" description_elem = prop.find(
if not href_text.endswith("/"): ".//{urn:ietf:params:xml:ns:caldav}calendar-description"
continue # Skip non-calendar resources )
description = description_elem.text if description_elem is not None else ""
# Extract calendar name from href color_elem = prop.find(".//{http://calendarserver.org/ns/}calendar-color")
calendar_name = href_text.rstrip("/").split("/")[-1] color = color_elem.text if color_elem is not None else "#1976D2"
if not calendar_name or calendar_name == self.username:
continue
# Get properties calendars.append(
propstat = response_elem.find(".//{DAV:}propstat") {
if propstat is None: "name": calendar_name,
continue "display_name": displayname,
"description": description,
"color": color,
"href": href_text,
}
)
prop = propstat.find(".//{DAV:}prop") logger.debug(f"Found {len(calendars)} calendars")
if prop is None: return calendars
continue
# Check if it's a calendar resource
resourcetype = prop.find(".//{DAV:}resourcetype")
is_calendar = (
resourcetype is not None
and resourcetype.find(".//{urn:ietf:params:xml:ns:caldav}calendar")
is not None
)
if not is_calendar:
continue
# Extract calendar properties
displayname_elem = prop.find(".//{DAV:}displayname")
displayname = (
displayname_elem.text
if displayname_elem is not None
else calendar_name
)
description_elem = prop.find(
".//{urn:ietf:params:xml:ns:caldav}calendar-description"
)
description = (
description_elem.text if description_elem is not None else ""
)
color_elem = prop.find(
".//{http://calendarserver.org/ns/}calendar-color"
)
color = color_elem.text if color_elem is not None else "#1976D2"
calendars.append(
{
"name": calendar_name,
"display_name": displayname,
"description": description,
"color": color,
"href": href_text,
}
)
logger.debug(f"Found {len(calendars)} calendars")
return calendars
except HTTPStatusError as e:
if e.response.status_code == 401:
logger.warning(
"Authentication failed for CalDAV - Calendar app may not be enabled for this user"
)
return []
elif e.response.status_code == 404:
logger.warning(
"CalDAV endpoint not found - Calendar app may not be installed"
)
return []
logger.error(f"HTTP error listing calendars: {e}")
raise e
except Exception as e:
logger.error(f"Unexpected error listing calendars: {e}")
raise e
async def get_calendar_events( async def get_calendar_events(
self, self,
@@ -180,55 +156,42 @@ class CalendarClient(BaseNextcloudClient):
"Accept": "application/xml", "Accept": "application/xml",
} }
try: response = await self._make_request(
response = await self._client.request( "REPORT", calendar_path, content=report_body, headers=headers
"REPORT", calendar_path, content=report_body, headers=headers )
)
response.raise_for_status()
# Parse XML response and extract events # Parse XML response and extract events
root = ET.fromstring(response.content) root = ET.fromstring(response.content)
events = [] events = []
for response_elem in root.findall(".//{DAV:}response"): for response_elem in root.findall(".//{DAV:}response"):
href = response_elem.find(".//{DAV:}href") href = response_elem.find(".//{DAV:}href")
if href is None: if href is None:
continue continue
propstat = response_elem.find(".//{DAV:}propstat") propstat = response_elem.find(".//{DAV:}propstat")
if propstat is None: if propstat is None:
continue continue
prop = propstat.find(".//{DAV:}prop") prop = propstat.find(".//{DAV:}prop")
if prop is None: if prop is None:
continue continue
calendar_data = prop.find( calendar_data = prop.find(".//{urn:ietf:params:xml:ns:caldav}calendar-data")
".//{urn:ietf:params:xml:ns:caldav}calendar-data" etag_elem = prop.find(".//{DAV:}getetag")
)
etag_elem = prop.find(".//{DAV:}getetag")
if calendar_data is not None and calendar_data.text: if calendar_data is not None and calendar_data.text:
event_data = self._parse_ical_event(calendar_data.text) event_data = self._parse_ical_event(calendar_data.text)
if event_data: if event_data:
event_data["href"] = href.text event_data["href"] = href.text
event_data["etag"] = ( event_data["etag"] = etag_elem.text if etag_elem is not None else ""
etag_elem.text if etag_elem is not None else "" events.append(event_data)
)
events.append(event_data)
if len(events) >= limit: if len(events) >= limit:
break break
logger.debug(f"Found {len(events)} events") logger.debug(f"Found {len(events)} events")
return events return events
except HTTPStatusError as e:
logger.error(f"HTTP error getting calendar events: {e}")
raise e
except Exception as e:
logger.error(f"Unexpected error getting calendar events: {e}")
raise e
async def create_event( async def create_event(
self, calendar_name: str, event_data: Dict[str, Any] self, calendar_name: str, event_data: Dict[str, Any]
@@ -246,26 +209,17 @@ class CalendarClient(BaseNextcloudClient):
"If-None-Match": "*", # Ensure we're creating, not updating "If-None-Match": "*", # Ensure we're creating, not updating
} }
try: response = await self._make_request(
response = await self._client.put( "PUT", event_path, content=ical_content, headers=headers
event_path, content=ical_content, headers=headers )
)
response.raise_for_status()
logger.debug(f"Created event {event_uid}") logger.debug(f"Created event {event_uid}")
return { return {
"uid": event_uid, "uid": event_uid,
"href": event_path, "href": event_path,
"etag": response.headers.get("etag", ""), "etag": response.headers.get("etag", ""),
"status_code": response.status_code, "status_code": response.status_code,
} }
except HTTPStatusError as e:
logger.error(f"HTTP error creating event: {e}")
raise e
except Exception as e:
logger.error(f"Unexpected error creating event: {e}")
raise e
async def update_event( async def update_event(
self, self,
@@ -303,10 +257,9 @@ class CalendarClient(BaseNextcloudClient):
headers["If-Match"] = etag headers["If-Match"] = etag
try: try:
response = await self._client.put( response = await self._make_request(
event_path, content=ical_content, headers=headers "PUT", event_path, content=ical_content, headers=headers
) )
response.raise_for_status()
logger.debug(f"Updated event {event_uid}") logger.debug(f"Updated event {event_uid}")
return { return {
@@ -329,8 +282,7 @@ class CalendarClient(BaseNextcloudClient):
event_path = f"{self._get_caldav_base_path()}/{calendar_name}/{event_filename}" event_path = f"{self._get_caldav_base_path()}/{calendar_name}/{event_filename}"
try: try:
response = await self._client.delete(event_path) response = await self._make_request("DELETE", event_path)
response.raise_for_status()
logger.debug(f"Deleted event {event_uid}") logger.debug(f"Deleted event {event_uid}")
return {"status_code": response.status_code} return {"status_code": response.status_code}
@@ -355,8 +307,7 @@ class CalendarClient(BaseNextcloudClient):
headers = {"Accept": "text/calendar"} headers = {"Accept": "text/calendar"}
try: try:
response = await self._client.get(event_path, headers=headers) response = await self._make_request("GET", event_path, headers=headers)
response.raise_for_status()
etag = response.headers.get("etag", "") etag = response.headers.get("etag", "")
event_data = self._parse_ical_event(response.text) event_data = self._parse_ical_event(response.text)
@@ -943,10 +894,9 @@ class CalendarClient(BaseNextcloudClient):
headers = {"Content-Type": "application/xml", "Depth": "0"} headers = {"Content-Type": "application/xml", "Depth": "0"}
response = await self._client.request( response = await self._make_request(
"MKCALENDAR", calendar_path, content=mkcol_body, headers=headers "MKCALENDAR", calendar_path, content=mkcol_body, headers=headers
) )
response.raise_for_status()
logger.debug(f"Created calendar: {calendar_name}") logger.debug(f"Created calendar: {calendar_name}")
return { return {
@@ -966,8 +916,7 @@ class CalendarClient(BaseNextcloudClient):
try: try:
calendar_path = f"{self._get_caldav_base_path()}/{calendar_name}/" calendar_path = f"{self._get_caldav_base_path()}/{calendar_name}/"
response = await self._client.delete(calendar_path) response = await self._make_request("DELETE", calendar_path)
response.raise_for_status()
logger.debug(f"Deleted calendar: {calendar_name}") logger.debug(f"Deleted calendar: {calendar_name}")
return {"status_code": response.status_code} return {"status_code": response.status_code}