Feature/add sharepoint service

Add a SharePoint service that can create new list items for approval checklists.

gso/services/sharepoint.py

+"""Sharepoint service used for creating new list items."""
+
+import asyncio
+
+from azure.identity.aio import CertificateCredential
+from msgraph import GraphServiceClient
+from msgraph.generated.models.field_value_set import FieldValueSet
+from msgraph.generated.models.list_item import ListItem
+from msgraph.generated.models.list_item_collection_response import ListItemCollectionResponse
+from msgraph.generated.models.site import Site
+from msgraph.generated.sites.item.lists.item.items.items_request_builder import ItemsRequestBuilder
+
+from gso.settings import load_oss_params
+
+
+class SharePointClient:
+    """A client for interaction with SharePoint lists."""
+
+    def __init__(self) -> None:
+        """Initialise a new SharePoint client."""
+        sp_params = load_oss_params().SHAREPOINT
+        _credentials = CertificateCredential(
+            tenant_id=sp_params.tenant_id,
+            client_id=sp_params.client_id,
+            certificate_path=sp_params.certificate_path,
+            password=sp_params.certificate_password,
+        )
+        self.client = GraphServiceClient(_credentials, sp_params.scopes)
+        self.site_id = sp_params.site_id
+        self.list_ids = sp_params.list_ids
+
+    def get_site(self) -> Site | None:
+        """Get the SharePoint site that this orchestrator connects to."""
+
+        async def _get_site() -> Site | None:
+            return await self.client.sites.by_site_id(self.site_id).get()
+
+        return asyncio.run(_get_site())
+
+    def get_list_items(self, list_name: str) -> ListItemCollectionResponse | None:
+        """Get list items from a given list in SharePoint.
+
+        :param str list_name: The name of the list.
+        """
+
+        async def _get_list_items() -> ListItemCollectionResponse | None:
+            query_params = ItemsRequestBuilder.ItemsRequestBuilderGetQueryParameters(expand=["fields"])
+            request_configuration = ItemsRequestBuilder.ItemsRequestBuilderGetRequestConfiguration(
+                query_parameters=query_params
+            )
+            return (
+                await self.client.sites.by_site_id(self.site_id)
+                .lists.by_list_id(self.list_ids[list_name])
+                .items.get(request_configuration=request_configuration)
+            )
+
+        return asyncio.run(_get_list_items())
+
+    def add_list_item(self, list_name: str, fields: dict[str, str]) -> str:
+        """Add a new entry to a SharePoint list.
+
+        :param str list_name: The name of the list.
+        :param dict[str, str] fields: Any pre-filled fields in the list item. Can be left empty.
+
+        :return str: The URL of the list in which a new item has been created.
+        """
+
+        async def _new_list_item() -> str:
+            request_body = ListItem(fields=FieldValueSet(additional_data=fields))
+            new_item = (
+                await self.client.sites.by_site_id(self.site_id)
+                .lists.by_list_id(self.list_ids[list_name])
+                .items.post(request_body)
+            )
+
+            #  Strip the last part of the URL, since we want the link to the list, not the list item.
+            return new_item.web_url.rsplit("/", 1)[0]
+
+        return asyncio.run(_new_list_item())