diff --git a/gso/services/sharepoint.py b/gso/services/sharepoint.py new file mode 100644 index 0000000000000000000000000000000000000000..4b4f421cfb2f2bd3fc754c3ab6ccba75300d95ca --- /dev/null +++ b/gso/services/sharepoint.py @@ -0,0 +1,62 @@ +"""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) + ) diff --git a/gso/settings.py b/gso/settings.py index 0e06192c56c6f57c1ac012984990787166869b18..cb09073029cc818070599a5893546e17b8de1a60 100644 --- a/gso/settings.py +++ b/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): diff --git a/requirements.txt b/requirements.txt index 25b297607030fdf534b44612cf6417a836e18e13..3463f0c52bfebe389377947e0ae1d39065e0fdc0 100644 --- a/requirements.txt +++ b/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