Skip to content
Snippets Groups Projects

Feature/add sharepoint service

Merged Karel van Klink requested to merge feature/add-sharepoint-service into develop
3 files
+ 76
4
Compare changes
  • Side-by-side
  • Inline
Files
3
+ 62
0
"""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)
)
Loading