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

make sharepoint client blocking

parent f81376b6
No related branches found
No related tags found
No related merge requests found
"""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
......@@ -27,26 +29,34 @@ class SPClient:
self.site_id = sp_params.site_id
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."""
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.
: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.
:param str list_name: The name of the list.
......@@ -54,12 +64,16 @@ class SPClient:
: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.
return new_item.web_url.rsplit("/", 1)[0]
async def _new_list_item():
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