Skip to content
Snippets Groups Projects
Verified Commit 9d1f2ea2 authored by Karel van Klink's avatar Karel van Klink :smiley_cat:
Browse files

make sharepoint client blocking

parent d29d3063
Branches
Tags
1 merge request!211Feature/add sharepoint service
"""Sharepoint service used for creating new list items.""" """Sharepoint service used for creating new list items."""
import asyncio
from azure.identity.aio import CertificateCredential from azure.identity.aio import CertificateCredential
from msgraph import GraphServiceClient from msgraph import GraphServiceClient
from msgraph.generated.models.field_value_set import FieldValueSet from msgraph.generated.models.field_value_set import FieldValueSet
...@@ -27,26 +29,34 @@ class SPClient: ...@@ -27,26 +29,34 @@ class SPClient:
self.site_id = sp_params.site_id self.site_id = sp_params.site_id
self.list_ids = sp_params.list_ids self.list_ids = sp_params.list_ids
async def get_site(self) -> Site | None: def get_site(self) -> Site | None:
"""Get the SharePoint site that this orchestrator connects to.""" """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: async def _get_site():
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. """Get list items from a given list in SharePoint.
:param str list_name: The name of the list. :param str list_name: The name of the list.
""" """
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)
)
async def add_list_item(self, list_name: str, fields: dict[str, str]) -> str: async def _get_list_items():
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. """Add a new entry to a SharePoint list.
:param str list_name: The name of the list. :param str list_name: The name of the list.
...@@ -54,12 +64,16 @@ class SPClient: ...@@ -54,12 +64,16 @@ class SPClient:
:return str: The URL of the list in which a new item has been created. :return str: The URL of the list in which a new item has been created.
""" """
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. async def _new_list_item():
return new_item.web_url.rsplit("/", 1)[0] 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())
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment