add-function

# src/sempy_labs/admin/_my_admin_function.py
# src/sempy_labs/lakehouse/_my_lakehouse_function.py
# src/sempy_labs/report/_my_report_function.py

import pandas as pd
from typing import Optional, List
from uuid import UUID
 
# Logging decorator from sempy
from sempy._utils._log import log
 
# Helper functions
from sempy_labs._helper_functions import (
    resolve_workspace_name_and_id,
    resolve_workspace_id,
    _base_api,
    _create_dataframe,
)
 
# Icons for user messages
import sempy_labs._icons as icons

@log
def my_new_function(
    item: str | UUID,
    workspace: Optional[str | UUID] = None,
    option: str = "default",
) -> pd.DataFrame:
    """
    Short description of what the function does.
 
    Extended description with more details about the function's behavior,
    use cases, and any important notes.
 
    This is a wrapper function for the following API: `API Name <https://learn.microsoft.com/rest/api/...>`_.
 
    Service Principal Authentication is supported (see `here <https://github.com/microsoft/semantic-link-labs/blob/main/notebooks/Service%20Principal.ipynb>`_ for examples).
 
    Parameters
    ----------
    item : str | uuid.UUID
        The name or ID of the item.
    workspace : str | uuid.UUID, default=None
        The Fabric workspace name or ID.
        Defaults to None which resolves to the workspace of the attached lakehouse
        or if no lakehouse attached, resolves to the workspace of the notebook.
    option : str, default="default"
        An option that controls function behavior.
 
    Returns
    -------
    pandas.DataFrame
        A pandas dataframe showing the results.
        Columns include: 'Column1', 'Column2', 'Column3'.
 
    Raises
    ------
    ValueError
        If the item does not exist.
    FabricHTTPException
        If the API request fails.
    """
 
    # Resolve workspace
    (workspace_name, workspace_id) = resolve_workspace_name_and_id(workspace)
 
    # Define result DataFrame structure
    columns = {
        "Column1": "string",
        "Column2": "string",
        "Column3": "int",
    }
    df = _create_dataframe(columns=columns)
 
    # Make API call
    responses = _base_api(
        request=f"/v1/workspaces/{workspace_id}/items",
        uses_pagination=True,
        client="fabric_sp",
    )
 
    # Process responses
    rows = []
    for r in responses:
        for item in r.get("value", []):
            rows.append({
                "Column1": item.get("id"),
                "Column2": item.get("name"),
                "Column3": item.get("count", 0),
            })
 
    if rows:
        df = pd.DataFrame(rows)
 
    return df

# src/sempy_labs/admin/__init__.py (example for admin submodule)
 
from ._my_admin_function import my_new_function
 
__all__ = [
    ...,
    "my_new_function",
]

from ._my_feature import my_new_function
 
__all__ = [
    ...,
    "my_new_function",
]

@log
def create_item(
    name: str,
    workspace: Optional[str | UUID] = None,
) -> None:
    """
    Creates a new item.
    ...
    """
    (workspace_name, workspace_id) = resolve_workspace_name_and_id(workspace)
 
    payload = {
        "displayName": name,
    }
 
    _base_api(
        request=f"/v1/workspaces/{workspace_id}/items",
        method="post",
        payload=payload,
        status_codes=[201, 202],
        client="fabric_sp",
    )
 
    print(
        f"{icons.green_dot} The '{name}' item has been successfully created "
        f"in the '{workspace_name}' workspace."
    )

@log
def delete_item(
    item: str | UUID,
    workspace: Optional[str | UUID] = None,
) -> None:
    """
    Deletes an item.
    ...
    """
    (workspace_name, workspace_id) = resolve_workspace_name_and_id(workspace)
    item_id = resolve_item_id(item=item, type="ItemType", workspace=workspace_id)
 
    _base_api(
        request=f"/v1/workspaces/{workspace_id}/items/{item_id}",
        method="delete",
        client="fabric_sp",
    )
 
    print(
        f"{icons.green_dot} The item has been successfully deleted "
        f"from the '{workspace_name}' workspace."
    )

@log
def long_running_operation(
    item: str | UUID,
    workspace: Optional[str | UUID] = None,
) -> dict:
    """
    Performs a long-running operation.
    ...
    """
    workspace_id = resolve_workspace_id(workspace)
    item_id = resolve_item_id(item=item, type="ItemType", workspace=workspace_id)
 
    # lro_return_json handles polling for completion
    result = _base_api(
        request=f"/v1/workspaces/{workspace_id}/items/{item_id}/operation",
        method="post",
        lro_return_json=True,
        client="fabric_sp",
    )
 
    return result

# tests/test_my_feature.py
 
import pytest
import pandas as pd
 
 
def test_my_new_function_returns_dataframe():
    """Test that my_new_function returns a DataFrame."""
    from sempy_labs import my_new_function
 
    # This might require mocking for unit tests
    result = my_new_function()
 
    assert isinstance(result, pd.DataFrame)
 
 
def test_my_new_function_with_workspace():
    """Test my_new_function with specific workspace."""
    from sempy_labs import my_new_function
 
    result = my_new_function(workspace="Test Workspace")
 
    assert isinstance(result, pd.DataFrame)