👌 IMPROVE: Add support for extra headers in components

Author: Samuel Guillemet

ms_python_client/api_client.py

@@ -1,13 +1,13 @@
 import logging
-from typing import Any, Dict, Mapping, Optional, TypeAlias
+from typing import Any, Mapping, Optional, TypeAlias
 
 from requests import RequestException, Response, Session
 from requests.adapters import HTTPAdapter
 from urllib3 import Retry
 
 logger = logging.getLogger("ms_python_client")
 
-_Headers: TypeAlias = Dict[str, str]
+_Headers: TypeAlias = Mapping[str, str]
 _Data: TypeAlias = Mapping[str, Any]
 
 
@@ -30,12 +30,12 @@ def build_headers(self, extra_headers: Optional[_Headers] = None) -> dict:
         """Create the headers for a request appending the ones in the params
 
         Args:
-            extra_headers (dict): Dict of headers that will be appended to the default ones
+            extra_headers (dict): Mapping of headers that will be appended to the default ones
 
         Returns:
             dict: All the headers
         """
-        headers: _Headers = {}
+        headers: dict[str, str] = {}
         if extra_headers:
             headers.update(extra_headers)
         return headers

ms_python_client/components/events/cern_events_component.py

@@ -28,7 +28,10 @@ def __init__(self, client: MSClientInterface) -> None:
         self.events_component = EventsComponent(client)
 
     def list_events(
-        self, user_id: str, parameters: Optional[Mapping[str, str]] = None
+        self,
+        user_id: str,
+        parameters: Optional[Mapping[str, str]] = None,
+        extra_headers: Optional[Mapping[str, str]] = None,
     ) -> dict:
         """List all the events of a user
 
@@ -39,34 +42,44 @@ def list_events(
         Returns:
             dict: The response of the request
         """
-        return self.events_component.list_events(user_id, parameters)
+        return self.events_component.list_events(user_id, parameters, extra_headers)
 
-    def get_event_by_indico_id(self, user_id: str, indico_id: str) -> dict:
+    def get_event_by_zoom_id(
+        self,
+        user_id: str,
+        zoom_id: str,
+        extra_headers: Optional[Mapping[str, str]] = None,
+    ) -> dict:
         """Get an event of a user
 
         Args:
-            indico_id (str): The event id
+            zoom_id (str): The event id
 
         Returns:
             dict: The response of the request
         """
-        parameters = {"$count": "true", "$filter": f"contains(subject,'{indico_id}')"}
-        response = self.events_component.list_events(user_id, parameters)
+        parameters = {"$count": "true", "$filter": f"contains(subject,'{zoom_id}')"}
+        response = self.events_component.list_events(user_id, parameters, extra_headers)
 
         count = response.get("@odata.count", 0)
         if count == 0:
-            raise NotFoundError(f"Event with indico id {indico_id} not found")
+            raise NotFoundError(f"Event with zoom id {zoom_id} not found")
 
         if count > 1:
             logger.warning(
-                "Found %s events with indico id %s. Returning the first one.",
+                "Found %s events with zoom id %s. Returning the first one.",
                 count,
-                indico_id,
+                zoom_id,
             )
 
         return response.get("value", [])[0]
 
-    def create_event(self, user_id: str, event: EventParameters) -> dict:
+    def create_event(
+        self,
+        user_id: str,
+        event: EventParameters,
+        extra_headers: Optional[Mapping[str, str]] = None,
+    ) -> dict:
         """Create an event for a user
 
         Args:
@@ -76,11 +89,14 @@ def create_event(self, user_id: str, event: EventParameters) -> dict:
         Returns:
             dict: The response of the request
         """
-        data = create_event_body(event)
-        return self.events_component.create_event(user_id, data)
-
-    def update_event_by_indico_id(
-        self, user_id: str, event: PartialEventParameters
+        json = create_event_body(event)
+        return self.events_component.create_event(user_id, json, extra_headers)
+
+    def update_event_by_zoom_id(
+        self,
+        user_id: str,
+        event: PartialEventParameters,
+        extra_headers: Optional[Mapping[str, str]] = None,
     ) -> dict:
         """Update an event for a user
 
@@ -91,19 +107,28 @@ def update_event_by_indico_id(
         Returns:
             dict: The response of the request
         """
-        data = create_partial_event_body(event)
-        event_id = self.get_event_by_indico_id(user_id, event["indico_event_id"])["id"]
-        return self.events_component.update_event(user_id, event_id, data)
-
-    def delete_event_by_indico_id(self, user_id: str, indico_id: str) -> None:
+        json = create_partial_event_body(event)
+        event_id = self.get_event_by_zoom_id(user_id, event["zoom_id"], extra_headers)[
+            "id"
+        ]
+        return self.events_component.update_event(
+            user_id, event_id, json, extra_headers
+        )
+
+    def delete_event_by_zoom_id(
+        self,
+        user_id: str,
+        zoom_id: str,
+        extra_headers: Optional[Mapping[str, str]] = None,
+    ) -> None:
         """Delete an event of a user
 
         Args:
             user_id (str): The user id
-            indico_id (str): The event id
+            zoom_id (str): The event id
 
         Returns:
             dict: The response of the request
         """
-        event_id = self.get_event_by_indico_id(user_id, indico_id)["id"]
-        self.events_component.delete_event(user_id, event_id)
+        event_id = self.get_event_by_zoom_id(user_id, zoom_id, extra_headers)["id"]
+        self.events_component.delete_event(user_id, event_id, extra_headers)

ms_python_client/components/events/events_component.py

@@ -8,7 +8,10 @@ def __init__(self, client: MSClientInterface) -> None:
         self.client = client
 
     def list_events(
-        self, user_id: str, parameters: Optional[Mapping[str, str]] = None
+        self,
+        user_id: str,
+        parameters: Optional[Mapping[str, str]] = None,
+        extra_headers: Optional[Mapping[str, str]] = None,
     ) -> dict:
         """List all the events of a user
 
@@ -19,10 +22,17 @@ def list_events(
             dict: The response of the request
         """
         api_path = f"/users/{user_id}/calendar/events"
-        response = self.client.make_get_request(api_path, parameters)
+        response = self.client.make_get_request(
+            api_path, parameters, extra_headers=extra_headers
+        )
         return response.json()
 
-    def get_event(self, user_id: str, event_id: str) -> dict:
+    def get_event(
+        self,
+        user_id: str,
+        event_id: str,
+        extra_headers: Optional[Mapping[str, str]] = None,
+    ) -> dict:
         """Get an event of a user
 
         Args:
@@ -33,10 +43,15 @@ def get_event(self, user_id: str, event_id: str) -> dict:
             dict: The response of the request
         """
         api_path = f"/users/{user_id}/calendar/events/{event_id}"
-        response = self.client.make_get_request(api_path)
+        response = self.client.make_get_request(api_path, extra_headers=extra_headers)
         return response.json()
 
-    def create_event(self, user_id: str, data: Mapping[str, Any]) -> dict:
+    def create_event(
+        self,
+        user_id: str,
+        json: Mapping[str, Any],
+        extra_headers: Optional[Mapping[str, str]] = None,
+    ) -> dict:
         """Create an event for a user
 
         Args:
@@ -47,11 +62,17 @@ def create_event(self, user_id: str, data: Mapping[str, Any]) -> dict:
             dict: The response of the request
         """
         api_path = f"/users/{user_id}/calendar/events"
-        response = self.client.make_post_request(api_path, data)
+        response = self.client.make_post_request(
+            api_path, json, extra_headers=extra_headers
+        )
         return response.json()
 
     def update_event(
-        self, user_id: str, event_id: str, data: Mapping[str, Any]
+        self,
+        user_id: str,
+        event_id: str,
+        json: Mapping[str, Any],
+        extra_headers: Optional[Mapping[str, str]] = None,
     ) -> dict:
         """Update an event for a user
 
@@ -64,10 +85,17 @@ def update_event(
             dict: The response of the request
         """
         api_path = f"/users/{user_id}/calendar/events/{event_id}"
-        response = self.client.make_patch_request(api_path, data)
+        response = self.client.make_patch_request(
+            api_path, json, extra_headers=extra_headers
+        )
         return response.json()
 
-    def delete_event(self, user_id: str, event_id: str) -> None:
+    def delete_event(
+        self,
+        user_id: str,
+        event_id: str,
+        extra_headers: Optional[Mapping[str, str]] = None,
+    ) -> None:
         """Delete an event of a user
 
         Args:
@@ -78,4 +106,4 @@ def delete_event(self, user_id: str, event_id: str) -> None:
             dict: The response of the request
         """
         api_path = f"/users/{user_id}/calendar/events/{event_id}"
-        self.client.make_delete_request(api_path)
+        self.client.make_delete_request(api_path, extra_headers=extra_headers)

ms_python_client/components/events/users_component.py

@@ -7,15 +7,20 @@ class UsersComponent:
     def __init__(self, client: MSClientInterface) -> None:
         self.client = client
 
-    def list_users(self, parameters: Optional[Mapping[str, str]] = None) -> dict:
+    def list_users(
+        self,
+        parameters: Optional[Mapping[str, str]] = None,
+        extra_headers: Optional[Mapping[str, str]] = None,
+    ) -> dict:
         """List all users
 
         Args:
             parameters (Optional[Mapping[str, str]], optional): Parameters for the request. Defaults to None.
+            extra_headers (Optional[Mapping[str, str]], optional): Additional headers for the request. Defaults to None.
 
         Returns:
             dict: The response of the request
         """
         api_path = "/users"
-        response = self.client.make_get_request(api_path, parameters)
+        response = self.client.make_get_request(api_path, parameters, extra_headers)
         return response.json()

ms_python_client/ms_api_client.py

@@ -1,7 +1,7 @@
 import atexit
 import logging
 import os
-from typing import Any, Dict, Mapping, Optional, TypeAlias
+from typing import Any, Mapping, Optional, TypeAlias
 
 import requests
 from msal import ConfidentialClientApplication, SerializableTokenCache
@@ -16,7 +16,7 @@
 logger = logging.getLogger("ms_python_client")
 
 _Data: TypeAlias = Mapping[str, Any]
-_Headers: TypeAlias = Dict[str, str]
+_Headers: TypeAlias = Mapping[str, str]
 
 
 class MSApiClient(MSClientInterface):

ms_python_client/ms_client_interface.py

@@ -1,13 +1,13 @@
 import logging
 from abc import ABC, abstractmethod
-from typing import Any, Dict, Mapping, Optional, TypeAlias
+from typing import Any, Mapping, Optional, TypeAlias
 
 import requests
 
 logger = logging.getLogger("ms_python_client")
 
 _Data: TypeAlias = Mapping[str, Any]
-_Headers: TypeAlias = Dict[str, str]
+_Headers: TypeAlias = Mapping[str, str]
 
 
 class MSClientInterface(ABC):