Implement SharePoint data extraction sources and configuration classes

[WangCHEN9]

Introduce classes and functions for extracting data from SharePoint lists and files, including configuration for file types and paths. This implementation supports both list and file data extraction into a DuckDB pipeline.

[WangCHEN9]

sources\sharepoint_pipeline.py tested locally OK

[anuunchin]

Hi @WangCHEN9, thanks for the contribution! Could add the tests you have written for the source?

btw, please rebase on master, some import failures were fixed which were causing ci errors

[WangCHEN9]

Hi @WangCHEN9, thanks for the contribution! Could add the tests you have written for the source?

btw, please rebase on master, some import failures were fixed which were causing ci errors

Added tests

[anuunchin]

It works well, but a few improvements are needed imo 🙂. Mainly around pagination (I know it was marked as a todo, but it should be fairly easy to add here), some edge cases that could cause runtime errors, and simplifying parts to better follow dlt conventions

If you have any questions, I’m happy to discuss further here or in the community Slack

[anuunchin]

Not specific to this file - would be great if you run make format and make lint, and address any issues they surface, otherwise the linter checks will fail

[anuunchin]

looks like FileType.SAV is missing here

[anuunchin]

This would be returning None, which is not reflected in the return type signature of the function - this is probably being raised by the linter as well

[anuunchin]

This is similar to the issue with the get_items_from_list function returning None

[anuunchin]

this seems redundant, as self.get_all_lists_in_site already handles the filtering

[anuunchin]

since get_items_from_list may currently return None, this will raise a NonType not iterable error

[anuunchin]

I think this will only work with pd.read_csv

[anuunchin]

Was the intention here to raise a tuple, not an error string?

[anuunchin]

wondering what the reason for doing manual incremental filtering here is - I'm thinking something like this is more idiomatic and readable:

@dlt.source(name="sharepoint_files", max_table_nesting=0)
def sharepoint_files(
    folder_path: str = dlt.config.value,
    table_name: str = dlt.config.value,
    file_type: str = dlt.config.value,
    file_name_prefix: Optional[str] = dlt.config.value,
    file_name_pattern: Optional[str] = dlt.config.value,
    pandas_kwargs: Optional[Dict] = dlt.config.value,
    credentials: SharepointCredentials = dlt.secrets.value,
):
    folder_path = validate_folder_path(folder_path)
    file_type_enum = FileType(file_type)
    
    client = SharepointClient(**credentials)
    client.connect()

    @dlt.resource(name=f"{table_name}__files", selected=False)
    def file_items(
        last_modified: dlt.sources.incremental[str] = dlt.sources.incremental(
            cursor_path="lastModifiedDateTime",
            initial_value="1970-01-01T00:00:00Z",
        ),
    ):
        for item in client.get_files_from_path(
            folder_path=folder_path,
            file_name_startswith=file_name_prefix or "",
            pattern=file_name_pattern,
        ):
            yield item

    @dlt.transformer(name=table_name)
    def file_data(file_item: Dict):
        file_io = client.get_file_bytes_io(file_item)
        pd_func = file_type_enum.get_pd_function()
        kwargs = pandas_kwargs or {}
        chunksize = kwargs.get("chunksize")
        
        if chunksize and file_type_enum == FileType.CSV:
            with pd_func(file_io, **kwargs) as reader:
                for chunk in reader:
                    yield chunk
        else:
            clean_kwargs = {k: v for k, v in kwargs.items() if k != "chunksize"}
            yield pd_func(file_io, **clean_kwargs)

    return file_items | file_data

[anuunchin]

same goes for the sharepoint_list source

@dlt.source(name="sharepoint_list", max_table_nesting=0)
def sharepoint_list(
    list_title: str = dlt.config.value,
    table_name: str = dlt.config.value,
    select: Optional[str] = dlt.config.value,
    credentials: SharepointCredentials = dlt.secrets.value,
):
    client = SharepointClient(**credentials)
    client.connect()

    @dlt.resource(name=table_name)
    def list_items():
        data = client.get_items_from_list(
            list_title=list_title,
            select=select,
        )
        yield from data

    return list_items

[anuunchin]

since the client has the paginator already defined, I think we can just go with something like

def get_items_from_list(self, list_title: str, select: Optional[str] = None) -> Iterator[Dict]:
    all_lists = self.get_all_lists_in_site()
    
    possible_list_titles = [x["displayName"] for x in all_lists]
    if list_title not in possible_list_titles:
        raise ValueError(
            f"List with title '{list_title}' not found in site {self.site_id}. "
            f"Available lists: {possible_list_titles}"
        )

    # Get the list ID
    list_id = next(
        x["id"] for x in all_lists if x["displayName"] == list_title
    )

    url = f"{self.graph_site_url}/lists/{list_id}/items?expand=fields"
    if select:
        url += f"(select={select})"
    
    # Paginate through all results
    item_count = 0
    for page in self.client.paginate(url):
        page.raise_for_status()
        for item in page.json().get("value", []):
            item_count += 1
            yield item.get("fields", {})
    
    logger.info(f"Got {item_count} items from list: {list_title}")

wdyt?