operations agents

src/sempy_labs/event_schema_set/__init__.py

@@ -0,0 +1,9 @@
+from ._items import (
+    list_event_schema_sets,
+    delete_event_schema_set,
+)
+
+__all__ = [
+    "list_event_schema_sets",
+    "delete_event_schema_set",
+]

src/sempy_labs/event_schema_set/_items.py

@@ -0,0 +1,84 @@
+import pandas as pd
+from typing import Optional
+from sempy_labs._helper_functions import (
+    _base_api,
+    _create_dataframe,
+    delete_item,
+    resolve_workspace_id,
+)
+from uuid import UUID
+from sempy._utils._log import log
+
+
+@log
+def list_event_schema_sets(workspace: Optional[str | UUID] = None) -> pd.DataFrame:
+    """
+    Returns a list of Event Schema Sets from the specified workspace.
+
+    This is a wrapper function for the following API: `Items - List Event Schema Sets <https://learn.microsoft.com/rest/api/fabric/eventschemaset/items/list-event-schema-sets>`_.
+    Parameters
+    ----------
+    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.
+
+    Returns
+    -------
+    pandas.DataFrame
+        A pandas dataframe showing the Event Schema Sets within a workspace.
+    """
+    workspace_id = resolve_workspace_id(workspace)
+
+    columns = {
+        "Event Schema Set Name": "str",
+        "Event Schema Set Id": "str",
+        "Description": "str",
+        "OneLake Root Path": "str",
+    }
+    df = _create_dataframe(columns=columns)
+
+    responses = _base_api(
+        request=f"v1/workspaces/{workspace_id}/eventSchemaSets",
+        client="fabric_sp",
+        uses_pagination=True,
+    )
+
+    rows = []
+
+    for r in responses:
+        for v in r.get("value", []):
+            row = {
+                "Event Schema Set Name": v.get("displayName"),
+                "Event Schema Set Id": v.get("id"),
+                "Description": v.get("description"),
+                "OneLake Root Path": v.get("properties", {}).get("oneLakeRootPath"),
+            }
+            rows.append(row)
+
+    if rows:
+        df = pd.DataFrame(rows, columns=list(columns.keys()))
+
+    return df
+
+
+@log
+def delete_event_schema_set(
+    event_schema_set: str | UUID, workspace: Optional[str | UUID] = None
+):
+    """
+    Deletes an Event Schema Set from the specified workspace.
+
+    This is a wrapper function for the following API: `Items - Delete Event Schema Set <https://learn.microsoft.com/rest/api/fabric/eventschemaset/items/delete-event-schema-set>`_.
+
+    Parameters
+    ----------
+    event_schema_set : str | uuid.UUID
+        The Event Schema Set name or ID to delete.
+    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.
+    """
+
+    delete_item(item=event_schema_set, type="EventSchemaSet", workspace=workspace)

src/sempy_labs/operations_agent/__init__.py

@@ -0,0 +1,9 @@
+from ._items import (
+    list_operations_agents,
+    delete_operations_agent,
+)
+
+__all__ = [
+    "list_operations_agents",
+    "delete_operations_agent",
+]

src/sempy_labs/operations_agent/_items.py

@@ -0,0 +1,82 @@
+import pandas as pd
+from typing import Optional
+from sempy_labs._helper_functions import (
+    _base_api,
+    _create_dataframe,
+    delete_item,
+    resolve_workspace_id,
+)
+from uuid import UUID
+from sempy._utils._log import log
+
+
+@log
+def list_operations_agents(workspace: Optional[str | UUID] = None) -> pd.DataFrame:
+    """
+    Returns a list of Operations Agents from the specified workspace.
+
+    This is a wrapper function for the following API: `Items - List Operations Agents <https://learn.microsoft.com/rest/api/fabric/operationsagent/items/list-operations-agents>`_.
+
+    Parameters
+    ----------
+    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.
+
+    Returns
+    -------
+    pandas.DataFrame
+        A pandas dataframe showing the Operations Agents within a workspace.
+    """
+    workspace_id = resolve_workspace_id(workspace)
+
+    columns = {
+        "Operations Agent Name": "str",
+        "Operations Agent Id": "str",
+        "Description": "str",
+        "State": "str",
+    }
+    df = _create_dataframe(columns=columns)
+
+    responses = _base_api(
+        request=f"v1/workspaces/{workspace_id}/OperationsAgents",
+        client="fabric_sp",
+        uses_pagination=True,
+    )
+
+    rows = []
+
+    for r in responses:
+        for v in r.get("value", []):
+            row = {
+                "Operations Agent Name": v.get("displayName"),
+                "Operations Agent Id": v.get("id"),
+                "Description": v.get("description"),
+                "State": v.get("properties", {}).get("state"),
+            }
+            rows.append(row)
+
+    return df
+
+
+@log
+def delete_operations_agent(
+    operations_agent: str | UUID, workspace: Optional[str | UUID] = None
+):
+    """
+    Deletes an Operations Agent from the specified workspace.
+
+    This is a wrapper function for the following API: `Items - Delete Operations Agent <https://learn.microsoft.com/rest/api/fabric/operationsagent/items/delete-operations-agent>`_.
+
+    Parameters
+    ----------
+    operations_agent : str | uuid.UUID
+        The Operations Agent name or ID to delete.
+    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.
+    """
+
+    delete_item(item=operations_agent, type="OperationsAgent", workspace=workspace)