Sharepoint

Sharepoint service used for creating new list items.

SharePointClient

A client for interaction with SharePoint lists.

Source code in gso/services/sharepoint.py

class SharePointClient:
    """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

    def get_site(self) -> Site | None:
        """Get the SharePoint site that this orchestrator connects to."""

        async def _get_site() -> Site | None:
            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.

        Args:
            list_name: The name of the list.
        """

        async def _get_list_items() -> ListItemCollectionResponse | None:
            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.

        Args:
            list_name: The name of the list.
            fields: Any pre-filled fields in the list item. Can be left empty.

        Returns:
            The URL of the list in which a new item has been created.
        """

        async def _new_list_item() -> str:
            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]  # type: ignore[union-attr]

        return asyncio.run(_new_list_item())

__init__()

Initialise a new SharePoint client.

Source code in gso/services/sharepoint.py

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

get_site()

Get the SharePoint site that this orchestrator connects to.

Source code in gso/services/sharepoint.py

def get_site(self) -> Site | None:
    """Get the SharePoint site that this orchestrator connects to."""

    async def _get_site() -> Site | None:
        return await self.client.sites.by_site_id(self.site_id).get()

    return asyncio.run(_get_site())

get_list_items(list_name)

Get list items from a given list in SharePoint.

Parameters:

Name	Type	Description	Default
list_name	str	The name of the list.	required

Source code in gso/services/sharepoint.py

def get_list_items(self, list_name: str) -> ListItemCollectionResponse | None:
    """Get list items from a given list in SharePoint.

    Args:
        list_name: The name of the list.
    """

    async def _get_list_items() -> ListItemCollectionResponse | None:
        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())

add_list_item(list_name, fields)

Add a new entry to a SharePoint list.

Parameters:

Name	Type	Description	Default
list_name	str	The name of the list.	required
fields	dict[str, str]	Any pre-filled fields in the list item. Can be left empty.	required

Returns:

Type	Description
str	The URL of the list in which a new item has been created.

Source code in gso/services/sharepoint.py

def add_list_item(self, list_name: str, fields: dict[str, str]) -> str:
    """Add a new entry to a SharePoint list.

    Args:
        list_name: The name of the list.
        fields: Any pre-filled fields in the list item. Can be left empty.

    Returns:
        The URL of the list in which a new item has been created.
    """

    async def _new_list_item() -> str:
        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]  # type: ignore[union-attr]

    return asyncio.run(_new_list_item())

prompt_sharepoint_checklist_url(checklist_url)

Prompt the operator with the checklist in SharePoint for approving a new subscription.

Source code in gso/services/sharepoint.py

@inputstep("Prompt for new SharePoint checklist", assignee=Assignee.SYSTEM)
def prompt_sharepoint_checklist_url(checklist_url: str) -> FormGenerator:
    """Prompt the operator with the checklist in SharePoint for approving a new subscription."""

    class SharepointPrompt(SubmitFormPage):
        model_config = ConfigDict(title="Complete new checklist")

        info_label_1: Label = f"A new checklist has been created at: {checklist_url}"
        info_label_2: Label = "Click proceed to finish the workflow."

    yield SharepointPrompt

    return {}