Add simple sharepoint client

gso/services/sharepoint.py

+"""Sharepoint service used for creating new list items."""
+
+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.sites.item.lists.item.items.items_request_builder import ItemsRequestBuilder
+from products import Site
+
+from gso.settings import load_oss_params
+
+
+class SPClient:
+    """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
+
+    async def get_site(self) -> Site | None:
+        """Get the SharePoint site that this orchestrator connects to."""
+        return await self.client.sites.by_site_id(self.site_id).get()
+
+    async 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.
+        """
+        query_params = ItemsRequestBuilder.ItemsRequestBuilderGetQueryParameters(
+            expand=["fields($select=Title,LinkTitle,CHECK_LIST_STATE,VERIFY_LIBRENMS)"],
+        )
+        request_configuration = ItemsRequestBuilder.ItemsRequestBuilderGetRequestConfiguration(
+            query_parameters=query_params
+        )
+        return (
+            await self.client.sites.by_site_id(self.site_id)
+            .lists.by_list_id(getattr(self.list_ids, list_name))
+            .items.get(request_configuration=request_configuration)
+        )
+
+    async def add_list_item(self, list_name: str, fields: dict[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.
+        """
+        request_body = ListItem(fields=FieldValueSet(additional_data=fields))
+        return (
+            await self.client.sites.by_site_id(self.site_id)
+            .lists.by_list_id(getattr(self.list_ids, list_name))
+            .items.post(request_body)
+        )

gso/settings.py

@@ -11,8 +11,8 @@ import os
 from pathlib import Path
 from typing import Annotated
 
-from pydantic import Field
-from pydantic_settings import BaseSettings
+from pydantic import BaseSettings, Field, HttpUrl
+from pydantic_forms.types import UUIDstr
 from typing_extensions import Doc
 
 logger = logging.getLogger(__name__)
@@ -161,8 +161,16 @@ class EmailParams(BaseSettings):
 class SharepointParams(BaseSettings):
     """Settings for different Sharepoint sites."""
 
-    # TODO: Stricter typing after Pydantic 2.x upgrade
-    checklist_site_url: str
+    client_id: UUIDstr
+    tenant_id: UUIDstr
+    certificate_path: str
+    certificate_password: str
+    site_id: UUIDstr
+    list_ids: dict[str, UUIDstr]
+    scopes: list[HttpUrl]
+    #:  .. deprecated :: 1.7
+    #:     Not used anymore, since this can be inferred from SharePoint :term:`API` responses.
+    checklist_site_url: HttpUrl | None
 
 
 class OSSParams(BaseSettings):

requirements.txt

@@ -5,6 +5,8 @@ pycountry==23.12.11
 pynetbox==7.3.3
 celery-redbeat==2.2.0
 celery==5.3.6
+azure-identity==1.16.0
+msgraph-sdk==1.2.0
 
 # Test and linting dependencies
 celery-stubs==0.1.3