diff --git a/README.md b/README.md index 3660db3..40d5587 100644 --- a/README.md +++ b/README.md @@ -19,7 +19,7 @@ pip install ms-python-client - Python >= 3.10 -## Usage +## How to configure the client variables to make API calls ### Defining your env variables @@ -31,12 +31,14 @@ Define the following variables in your `env` or your `.env` file: #### For testing purposes -For testing purposes, you can use the following values: +For testing purposes, you can use the following value: - MS_ACCESS_TOKEN This token could be obtained from the [Microsoft Graph Explorer](https://developer.microsoft.com/en-us/graph/graph-explorer) by clicking on the `Sign in with Microsoft` button and then clicking on the `Access Token` tab. +## Usage + ### Initialize the MSApiClient from environment variables ```python @@ -108,7 +110,7 @@ setup_logs(log_level=logging.DEBUG) 4. update an event 5. delete an event -## CERN specific endpoints +## CERN specific usage Instead of using the `MSApiClient` class, you can use the `CERNMSApiClient` class, which is a subclass of the `MSApiClient` class. This class will provide you some more utilities but it will only work for CERN users (obviously). @@ -117,10 +119,12 @@ This will be used in the context of synchronizing the events between the CERN In ### How to initialize the CERNMSApiClient +Follow the [How to configure the client variables to make API calls](#how-to-configure-the-client-variables-to-make-api-calls) section and then: + ```python from ms_python_client.cern_ms_api_client import CERNMSApiClient -cern_ms_client = CERNMSApiClient.init_from_env() +cern_ms_client = CERNMSApiClient.init_from_dotenv() ``` ### Available endpoints @@ -135,18 +139,14 @@ cern_ms_client = CERNMSApiClient.init_from_env() You will find useful the `EventParameters` and `PartialEventParameters` classes, which will help you to create the events. -**indico_event_id** is the id of the event in the indico system which is mandatory to create an event. - -**USER_ID** is the email of the Zoom Room. +- `INDICO_EVENT_ID` is the id of the event in the indico system which is **mandatory** to create an event. +- `USER_ID` is the email of the Zoom Room. ```python -from ms_python_client.cern_ms_api_client import ( - CERNMSApiClient, - EventParameters, - PartialEventParameters, - ) +from ms_python_client.cern_ms_api_client import CERNMSApiClient +from ms_python_client.utils.event_generator import (EventParameters, PartialEventParameters) -cern_ms_client = CERNMSApiClient.init_from_env() +cern_ms_client = CERNMSApiClient.init_from_dotenv() USER_ID = os.getenv("USER_ID") # Which is the email of the Zoom Room INDICO_EVENT_ID = os.getenv("INDICO_EVENT_ID") diff --git a/ms_python_client/cern_ms_api_client.py b/ms_python_client/cern_ms_api_client.py new file mode 100644 index 0000000..ff18c22 --- /dev/null +++ b/ms_python_client/cern_ms_api_client.py @@ -0,0 +1,45 @@ +from typing import Optional + +from ms_python_client.api_client import ApiClient +from ms_python_client.components.events.cern_events_component import ( + CERNEventsComponents, +) +from ms_python_client.ms_api_client import MSApiClient +from ms_python_client.utils import init_from_env + + +class CERNMSApiClient(MSApiClient): + def __init__( + self, + account_id: str, + client_id: str, + client_secret: str, + api_endpoint: str = "https://graph.microsoft.com/v1.0", + use_path: Optional[str] = None, + ): + super().__init__(account_id, client_id, client_secret, api_endpoint, use_path) + self.api_client = ApiClient(api_base_url=api_endpoint) + self.init_components() + + def init_components(self): + # Add all the new components here + self.events = CERNEventsComponents(self) + + @staticmethod + def init_from_dotenv( + custom_dotenv=".env", use_path: Optional[str] = None + ) -> "CERNMSApiClient": + init_from_env.init_from_dotenv(custom_dotenv) + ms_client = CERNMSApiClient.init_from_env(use_path=use_path) + return ms_client + + @staticmethod + def init_from_env(use_path: Optional[str] = None) -> "CERNMSApiClient": + values = init_from_env.init_from_env(use_path) + ms_client = CERNMSApiClient( + values["account_id"], + values["client_id"], + values["client_secret"], + use_path=values["use_path"], + ) + return ms_client diff --git a/ms_python_client/components/events/cern_events_component.py b/ms_python_client/components/events/cern_events_component.py new file mode 100644 index 0000000..20cfd66 --- /dev/null +++ b/ms_python_client/components/events/cern_events_component.py @@ -0,0 +1,109 @@ +import logging +from typing import Mapping, Optional + +from ms_python_client.components.events.events_component import EventsComponent +from ms_python_client.ms_client_interface import MSClientInterface +from ms_python_client.utils.event_generator import ( + EventParameters, + PartialEventParameters, + create_event_body, + create_partial_event_body, +) + +logger = logging.getLogger("ms_python_client") + + +class NotFoundError(Exception): + """Execption raised when an event is not found + + Args: + Exception (Exception): The base exception + """ + + +class CERNEventsComponents: + """CERN Events component""" + + def __init__(self, client: MSClientInterface) -> None: + self.events_component = EventsComponent(client) + + def list_events( + self, user_id: str, parameters: Optional[Mapping[str, str]] = None + ) -> dict: + """List all the events of a user + + Args: + user_id (str): The user id + parameters (dict): Optional parameters for the request + + Returns: + dict: The response of the request + """ + return self.events_component.list_events(user_id, parameters) + + def get_event_by_indico_id(self, user_id: str, indico_id: str) -> dict: + """Get an event of a user + + Args: + indico_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) + + count = response.get("@odata.count", 0) + if count == 0: + raise NotFoundError(f"Event with indico id {indico_id} not found") + + if count > 1: + logger.warning( + "Found %s events with indico id %s. Returning the first one.", + count, + indico_id, + ) + + return response.get("value", [])[0] + + def create_event(self, user_id: str, event: EventParameters) -> dict: + """Create an event for a user + + Args: + user_id (str): The user id + event (EventParameters): The event data + + 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 + ) -> dict: + """Update an event for a user + + Args: + user_id (str): The user id + event (EventParameters): The event parameters + + 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: + """Delete an event of a user + + Args: + user_id (str): The user id + indico_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) diff --git a/ms_python_client/utils/event_generator.py b/ms_python_client/utils/event_generator.py new file mode 100644 index 0000000..f95008a --- /dev/null +++ b/ms_python_client/utils/event_generator.py @@ -0,0 +1,162 @@ +import datetime +from typing import TypedDict + + +class BaseEventParameters(TypedDict): + """Base parameters for creating an event + + Args: + indico_event_id (str): The indico event id + """ + + indico_event_id: str + + +class OptionalTimezone(TypedDict, total=False): + """Optional timezone parameter for creating an event + + Args: + timezone (str): The timezone of the event + """ + + timezone: str + + +class EventParameters(BaseEventParameters, OptionalTimezone): + """Parameters for creating an event + + Args: + zoom_url (str): The Zoom URL for the event + subject (str): The subject of the event + start_time (str): The start time of the event in **ISO format** + end_time (str): The end time of the event in **ISO format** + """ + + zoom_url: str + subject: str + start_time: str + end_time: str + + +class PartialEventParameters(BaseEventParameters, OptionalTimezone, total=False): + """Parameters for updating an event + + Args: + zoom_url (str): The Zoom URL for the event + subject (str): The subject of the event + start_time (str): The start time of the event in **ISO format** + end_time (str): The end time of the event in **ISO format** + """ + + zoom_url: str + subject: str + start_time: str + end_time: str + + +def create_event_body(event_parameters: EventParameters) -> dict: + """Creates an event from the given parameters + + Args: + event_parameters (EventParameters): The parameters of the event + + Returns: + Event: The event + """ + + timezone = event_parameters.get("timezone", "Europe/Zurich") + + return { + "subject": f"[{event_parameters['indico_event_id']}] {event_parameters['subject']}", + "body": { + "contentType": "text", + "content": f"Zoom URL: {event_parameters['zoom_url']}", + }, + "start": { + "dateTime": datetime.datetime.fromisoformat( + event_parameters["start_time"] + ).isoformat(), + "timeZone": timezone, + }, + "end": { + "dateTime": datetime.datetime.fromisoformat( + event_parameters["end_time"] + ).isoformat(), + "timeZone": timezone, + }, + "location": { + "displayName": event_parameters["zoom_url"], + "locationType": "default", + "uniqueIdType": "private", + "uniqueId": event_parameters["zoom_url"], + }, + "attendees": [], + "allowNewTimeProposals": False, + "isOnlineMeeting": True, + "onlineMeetingProvider": "unknown", + "onlineMeetingUrl": event_parameters["zoom_url"], + } + + +def create_partial_event_body(event_parameters: PartialEventParameters) -> dict: + """Updates an event from the given parameters + + Args: + event_parameters (PartialEventParameters): The parameters of the event + + Returns: + Event: The event + """ + event = {} + + timezone = event_parameters.get("timezone", "Europe/Zurich") + + if "zoom_url" in event_parameters: + event.update( + { + "body": { + "contentType": "text", + "content": f"Zoom URL: {event_parameters['zoom_url']}", + }, + "location": { + "displayName": event_parameters["zoom_url"], + "locationType": "default", + "uniqueIdType": "private", + "uniqueId": event_parameters["zoom_url"], + }, + "onlineMeetingUrl": event_parameters["zoom_url"], + } + ) + + if "subject" in event_parameters: + event.update( + { + "subject": f"[{event_parameters['indico_event_id']}] {event_parameters['subject']}" + } + ) + + if "start_time" in event_parameters: + event.update( + { + "start": { + "dateTime": datetime.datetime.fromisoformat( + event_parameters["start_time"] + ).isoformat(), + "timeZone": timezone, + } + } + ) + + if "end_time" in event_parameters: + event.update( + { + "end": { + "dateTime": datetime.datetime.fromisoformat( + event_parameters["end_time"] + ).isoformat(), + "timeZone": timezone, + } + } + ) + + return event diff --git a/pyproject.toml b/pyproject.toml index 68f60ed..a52132b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "ms-python-client" -version = "0.1.1" +version = "0.2.0" exclude = ["tests*", "example*", ".github*", ".git*", ".vscode*"] description = "This package is used to interact with the microsoft graph API" authors = ["Samuel Guillemet "] diff --git a/tests/ms_python_client/components/events/test_cern_events_component.py b/tests/ms_python_client/components/events/test_cern_events_component.py new file mode 100644 index 0000000..52c8061 --- /dev/null +++ b/tests/ms_python_client/components/events/test_cern_events_component.py @@ -0,0 +1,124 @@ +import pytest +import responses + +from ms_python_client.cern_ms_api_client import CERNMSApiClient +from ms_python_client.components.events.cern_events_component import ( + CERNEventsComponents, + NotFoundError, +) +from ms_python_client.utils.event_generator import ( + EventParameters, + PartialEventParameters, +) +from tests.ms_python_client.base_test_case import TEST_API_ENDPOINT, BaseTest, mock_msal + + +class TestEventsComponent(BaseTest): + @mock_msal() + def setUp(self) -> None: + cern_ms_client = CERNMSApiClient( + "account_id", "client_id", "client_secret", api_endpoint=TEST_API_ENDPOINT + ) + self.events_component = CERNEventsComponents(cern_ms_client) + return super().setUp() + + @responses.activate + def test_list_events(self): + responses.add( + responses.GET, + "http://localhost/users/user_id/calendar/events", + json={"response": "ok"}, + status=200, + ) + events_list = self.events_component.list_events("user_id") + assert events_list["response"] == "ok" + + @responses.activate + def test_get_event_by_indico_id_0(self): + responses.add( + responses.GET, + "http://localhost/users/user_id/calendar/events", + json={"@odata.count": 0}, + status=200, + ) + with pytest.raises(NotFoundError): + self.events_component.get_event_by_indico_id("user_id", "indico_id") + + @responses.activate + def test_get_event_by_indico_id_1(self): + responses.add( + responses.GET, + "http://localhost/users/user_id/calendar/events", + json={ + "@odata.count": 2, + "value": [{"subject": "indico_id_1"}, {"subject": "indico_id_2"}], + }, + status=200, + ) + result = self.events_component.get_event_by_indico_id("user_id", "indico_id_1") + assert result["subject"] == "indico_id_1" + + @responses.activate + def test_create_event(self): + responses.add( + responses.POST, + "http://localhost/users/user_id/calendar/events", + json={"response": "ok"}, + status=200, + ) + event_parameters = EventParameters( + zoom_url="https://zoom.us/j/1234567890", + indico_event_id="1234567890", + subject="Test Event", + start_time="2021-01-01T00:00:00", + end_time="2021-01-01T01:00:00", + ) + event = self.events_component.create_event("user_id", event_parameters) + assert event["response"] == "ok" + + @responses.activate + def test_update_event(self): + responses.add( + responses.PATCH, + "http://localhost/users/user_id/calendar/events/event_id", + json={"response": "ok"}, + status=200, + ) + responses.add( + responses.GET, + "http://localhost/users/user_id/calendar/events", + json={ + "@odata.count": 1, + "value": [{"id": "event_id", "subject": "Test Event"}], + }, + status=200, + ) + event_parameters = PartialEventParameters( + zoom_url="https://zoom.us/j/1234567890", + indico_event_id="1234567890", + subject="Test Event", + start_time="2021-01-01T00:00:00", + end_time="2021-01-01T01:00:00", + ) + event = self.events_component.update_event_by_indico_id( + "user_id", event_parameters + ) + assert event["response"] == "ok" + + @responses.activate + def test_delete_event(self): + responses.add( + responses.DELETE, + "http://localhost/users/user_id/calendar/events/event_id", + status=204, + ) + responses.add( + responses.GET, + "http://localhost/users/user_id/calendar/events", + json={ + "@odata.count": 1, + "value": [{"id": "event_id", "subject": "Test Event"}], + }, + status=200, + ) + self.events_component.delete_event_by_indico_id("user_id", "indico_id") diff --git a/tests/ms_python_client/test_cern_ms_api_client.py b/tests/ms_python_client/test_cern_ms_api_client.py new file mode 100644 index 0000000..11bab63 --- /dev/null +++ b/tests/ms_python_client/test_cern_ms_api_client.py @@ -0,0 +1,21 @@ +import os + +from ms_python_client.cern_ms_api_client import CERNMSApiClient +from tests.ms_python_client.base_test_case import BaseTest, mock_msal + + +class TestCERNMSApiClientInit(BaseTest): + @mock_msal() + def test_init_from_env(self): + os.environ["MS_ACCOUNT_ID"] = "aaa" + os.environ["MS_CLIENT_ID"] = "bbb" + os.environ["MS_CLIENT_SECRET"] = "ccc" + + client = CERNMSApiClient.init_from_env() + + assert client is not None + + @mock_msal() + def test_init_from_dotenv(self): + client = CERNMSApiClient.init_from_dotenv(custom_dotenv=self.env_file) + assert client is not None diff --git a/tests/ms_python_client/utils/test_event_generator.py b/tests/ms_python_client/utils/test_event_generator.py new file mode 100644 index 0000000..06b5c65 --- /dev/null +++ b/tests/ms_python_client/utils/test_event_generator.py @@ -0,0 +1,101 @@ +from ms_python_client.utils.event_generator import ( + EventParameters, + PartialEventParameters, + create_event_body, + create_partial_event_body, +) + + +def test_create_event_body(): + parameters = EventParameters( + zoom_url="https://zoom.us/j/1234567890", + indico_event_id="1234567890", + subject="Test Event", + start_time="2021-01-01T00:00:00", + end_time="2021-01-01T01:00:00", + ) + + result = create_event_body(parameters) + + assert result == { + "subject": "[1234567890] Test Event", + "body": { + "contentType": "text", + "content": "Zoom URL: https://zoom.us/j/1234567890", + }, + "start": {"dateTime": "2021-01-01T00:00:00", "timeZone": "Europe/Zurich"}, + "end": {"dateTime": "2021-01-01T01:00:00", "timeZone": "Europe/Zurich"}, + "location": { + "displayName": "https://zoom.us/j/1234567890", + "locationType": "default", + "uniqueIdType": "private", + "uniqueId": "https://zoom.us/j/1234567890", + }, + "attendees": [], + "allowNewTimeProposals": False, + "isOnlineMeeting": True, + "onlineMeetingProvider": "unknown", + "onlineMeetingUrl": "https://zoom.us/j/1234567890", + } + + +def test_create_partial_event_body_0(): + parameters = PartialEventParameters( + indico_event_id="1234567890", + ) + + result = create_partial_event_body(parameters) + + assert not result + + +def test_create_partial_event_body_1(): + parameters = PartialEventParameters( + indico_event_id="1234567890", + zoom_url="https://zoom.us/j/1234567890", + ) + + result = create_partial_event_body(parameters) + + assert result == { + "body": { + "contentType": "text", + "content": "Zoom URL: https://zoom.us/j/1234567890", + }, + "location": { + "displayName": "https://zoom.us/j/1234567890", + "locationType": "default", + "uniqueIdType": "private", + "uniqueId": "https://zoom.us/j/1234567890", + }, + "onlineMeetingUrl": "https://zoom.us/j/1234567890", + } + + +def test_create_partial_event_body_2(): + parameters = PartialEventParameters( + indico_event_id="1234567890", + subject="Test Event", + ) + + result = create_partial_event_body(parameters) + + assert result == { + "subject": "[1234567890] Test Event", + } + + +def test_create_partial_event_body_3(): + parameters = PartialEventParameters( + indico_event_id="1234567890", + start_time="2021-01-01T00:00:00", + end_time="2021-01-01T01:00:00", + timezone="Europe/Zurich", + ) + + result = create_partial_event_body(parameters) + + assert result == { + "start": {"dateTime": "2021-01-01T00:00:00", "timeZone": "Europe/Zurich"}, + "end": {"dateTime": "2021-01-01T01:00:00", "timeZone": "Europe/Zurich"}, + }