proxied through sdk

Author: geekwhocodes

poetry.lock

@@ -30,6 +30,7 @@ grpcio = "^1.70.0"
 grpcio-status = "^1.60.1"
 pandas = "^2.2.0"
 ipykernel = "^6.29.5"
+markdown = "^3.7"
 
 [build-system]
 requires = ["poetry-core>=2.0.0,<3.0.0"]

pyproject.toml

@@ -14,7 +14,6 @@ def __init__(self, async_gen: AsyncGenerator[Any, None]):
 
         Args:
             async_gen (AsyncGenerator): The async generator yielding results.
-            max_pages (int): Maximum allowed pages to fetch (prevents infinite loops).
         """
         self.async_gen = async_gen
         self.loop = self._get_event_loop()
@@ -56,3 +55,68 @@ def __iter__(self) -> Iterator:
         """Returns the synchronous iterator."""
         return self.iterator
 
+
+import asyncio
+from typing import AsyncGenerator, Iterator, Any
+
+class AsyncToSyncIteratorV2:
+    """
+    Converts an async generator into a synchronous iterator while ensuring proper event loop handling.
+    """
+
+    def __init__(self, async_gen: AsyncGenerator[Any, None]):
+        """
+        Initializes the iterator by consuming an async generator synchronously.
+
+        Args:
+            async_gen (AsyncGenerator): The async generator yielding results.
+        """
+        self.async_gen = async_gen
+        self.iterator = self._to_iterator()
+
+    def _to_iterator(self) -> Iterator:
+        """
+        Ensures that the async generator is consumed using the correct event loop.
+        Uses streaming (does not load all results into memory).
+        """
+        try:
+            loop = asyncio.get_running_loop()
+            return self._sync_generator(loop)  # Works inside Jupyter
+        except RuntimeError:
+            return iter(asyncio.run(self._collect_results()))  # Works in scripts
+
+    def _sync_generator(self, loop: asyncio.AbstractEventLoop) -> Iterator:
+        """
+        Streams async results into a sync generator while inside a running event loop.
+        """
+        queue = asyncio.Queue()
+
+        async def _producer():
+            """Fills the queue with async results."""
+            async for item in self.async_gen:
+                await queue.put(item)
+            await queue.put(None)  # Sentinel to signal completion
+
+        async def _consumer():
+            """Yields items from the queue in sync mode."""
+            task = loop.create_task(_producer())
+            while True:
+                item = await queue.get()
+                if item is None:
+                    break
+                yield item
+            await task  # Ensure producer task completes
+
+        return iter(loop.run_until_complete(self._collect_results()))
+
+    async def _collect_results(self):
+        """Collects async generator results into a list (safe for asyncio.run)."""
+        return [item async for item in self.async_gen]
+
+    def __iter__(self) -> Iterator:
+        """Returns the synchronous iterator."""
+        return self.iterator
+
+    def __next__(self) -> Any:
+        """Returns the next item from the iterator."""
+        return next(self.iterator)

src/source_msgraph/async_interator.py

@@ -0,0 +1,105 @@
+from msgraph import GraphServiceClient
+from kiota_abstractions.base_request_configuration import RequestConfiguration
+from msgraph.generated.models.o_data_errors.o_data_error import ODataError
+from azure.identity import ClientSecretCredential
+from source_msgraph.async_interator import AsyncToSyncIterator, AsyncToSyncIteratorV2
+from source_msgraph.models import ConnectorOptions
+from source_msgraph.utils import get_python_schema, to_json, to_pyspark_schema
+
+class GraphClient:
+    def __init__(self, options: ConnectorOptions):
+        """
+        Initializes the fetcher with the Graph client, resource path, and query parameters.
+
+
+        :param options: Connector options.
+        """
+        credentials = ClientSecretCredential(options.tenant_id, options.client_id, options.client_secret)
+        self.graph_client = GraphServiceClient(credentials=credentials)
+        self.options: ConnectorOptions = options
+
+
+    async def fetch_data(self):
+        """
+        Fetches data from Microsoft Graph using the dynamically built request.
+        Handles pagination automatically.
+        """
+        query_parameters_cls = self.options.resource.get_query_parameters_cls()
+
+        if query_parameters_cls:
+            try:
+                query_parameters_instance = query_parameters_cls()  # Ensure it can be instantiated without arguments
+            except TypeError as e:
+                raise ValueError(f"Failed to instantiate {query_parameters_cls.__name__}: {e}")
+
+            if self.options.resource.query_params:
+                for k, v in self.options.resource.query_params.items():
+                    k = k.removeprefix("%24")
+                    if hasattr(query_parameters_instance, k):
+                        setattr(query_parameters_instance, k, v)  # Set attributes dynamically
+                    else:
+                        raise AttributeError(f"{query_parameters_cls.__name__} has no attribute '{k}'")
+                
+        request_configuration = RequestConfiguration(
+            query_parameters=query_parameters_instance
+        )
+        
+        try:
+            builder = self.options.resource.get_request_builder_cls()(self.graph_client.request_adapter, self.options.resource.resource_params)
+            items = await builder.get(request_configuration=request_configuration)
+            while True:
+                print("Page fetched....")
+                for item in items.value:
+                    yield item
+                if not items.odata_next_link:
+                    break
+                items = await builder.with_url(items.odata_next_link).get()
+
+        except ODataError as e:
+            raise Exception(f"Graph API Error: {e.error.message}")
+
+
+def iter_records(options: ConnectorOptions):
+    """
+        Iterates over records from the Microsoft Graph API.
+
+        :param options: Connector options containing authentication credentials and resource details.
+        :return: A synchronous iterator over the fetched data.
+        :raises ValueError: If any required credentials or resource parameters are missing.
+        :raises GraphAPIError: If the API request fails.
+    """
+    fetcher = GraphClient(options)
+    async_gen = fetcher.fetch_data()
+    return AsyncToSyncIterator(async_gen)
+
+import json
+from typing import Dict, Any
+from dataclasses import asdict
+
+def get_resource_schema(options: ConnectorOptions) -> Dict[str, Any]:
+    """
+    Retrieves the schema of a Microsoft Graph API resource by fetching a single record.
+
+    :param options: Connector options containing authentication credentials and resource details.
+    :return: A dictionary representing the schema of the resource.
+    :raises ValueError: If no records are found or if required options are missing.
+    :raises GraphAPIError: If the API request fails.
+    """
+    fetcher = GraphClient(options)
+    async_gen = fetcher.fetch_data()
+
+    try:
+        record = next(AsyncToSyncIteratorV2(async_gen), None)
+        if not record:
+            raise ValueError(f"No records found for resource: {options.resource.resource_name}")
+        record = to_json(record)
+        schema = to_pyspark_schema(get_python_schema(record))
+        return record, schema
+    
+    except StopIteration:
+        raise ValueError(f"No records available for {options.resource.resource_name}")
+
+# Example usage
+# options = ConnectorOptions(...)
+# schema = get_resource_schema(options)
+# print(json.dumps(schema, indent=2))

src/source_msgraph/client.py

@@ -0,0 +1,2 @@
+# Base generated package for Microsoft Graph SDK
+MSGRAPH_SDK_PACKAGE = "msgraph.generated"

src/source_msgraph/constants.py

@@ -0,0 +1,84 @@
+import os
+from urllib.parse import unquote
+from source_msgraph.models import BaseResource
+from source_msgraph.resources import RESOURCE_CONFIGS
+
+def generate_markdown(resource: BaseResource) -> str:
+    """
+    Generates sophisticated markdown documentation for a given BaseResource.
+    """
+    md_content = [f"# {resource.name.capitalize()} Resource", ""]
+    md_content.append(f"**Resource Name:** `{resource.name.lower()}`")
+    
+    
+    md_content.append("\n## Overview")
+    md_content.append(f"The `{resource.name}` resource provides a structured way to interact with Microsoft Graph API.")
+    md_content.append("This resource supports operations such as retrieval and filtering of data.")
+    
+    md_content.append("\n## Resource Parameters")
+    if len(resource.resource_params.keys()) > 0:
+        md_content.append("| Parameter | Type | Required | Description |")
+        md_content.append("|-----------|------|----------|-------------|")
+        for param in resource.resource_params or {}:
+            md_content.append(f"| `{unquote(param)}` | `str` | ✅ | Required path parameter for resource access. |")
+    else:
+        md_content.append(f"> No parameters required for `{resource.name.lower()}` resource.")
+
+
+    md_content.append("\n## Query Parameters")
+    if len(resource.query_params.keys()) > 0:
+        md_content.append("| Parameter | Type | Required | Description |")
+        md_content.append("|-----------|------|----------|-------------|")
+        for param in resource.query_params or {}:
+            md_content.append(f"| `{unquote(param)}` | `str` | ❌ | Optional query parameter to refine the API request. |")
+    else:
+        md_content.append(f">> No query parameters are required for `{resource.name.lower()}` resource.")
+
+    md_content.append("---")
+    
+    md_content.append("Tip: Please refer [Microsoft Graph API]() documentation if you don't see a field. This can be resolved by provising `expand` option.")
+
+    md_content.append("\n## Example Usage")
+    md_content.append("```python")
+    md_content.append("from source_msgraph.source import MSGraphDataSource")
+    md_content.append("spark.dataSource.register(MSGraphDataSource)")
+    md_content.append("")
+    md_content.append("# Read data using Microsoft Graph")
+    md_content.append("df = spark.read.format(\"msgraph\") ")
+    md_content.append("    .option(\"tenant_id\", tenant_id)")
+    md_content.append("    .option(\"client_id\", client_id)")
+    md_content.append("    .option(\"client_secret\", client_secret)")
+    md_content.append(f"    .option(\"resource\", \"{resource.name}\")")
+    for param in resource.resource_params or {}:
+        md_content.append(f"    .option(\"{param}\", \"<value>\")")
+    for param in resource.query_params or {}:
+        md_content.append(f"    .option(\"{param}\", \"<value>\")")
+    md_content.append("    .schema(\"id string, eTag string\")")
+    md_content.append("    .load()")
+    md_content.append("")
+    md_content.append("df.show()")
+    md_content.append("```")
+    
+    return "\n".join(md_content)
+
+def generate_docs(output_dir: str = "docs"):
+    """
+    Generates sophisticated markdown documentation for all configured resources.
+    """
+    os.makedirs(output_dir, exist_ok=True)
+    
+    for config in RESOURCE_CONFIGS:
+        resource = BaseResource(
+            name=config["name"],
+            resource_name=config["resource_name"],
+            request_builder_module=config["request_builder_module"]
+        )
+        
+        md_content = generate_markdown(resource)
+        file_path = os.path.join(output_dir, f"{resource.name}.md")
+        with open(file_path, "w", encoding="utf-8") as f:
+            f.write(md_content)
+        print(f"Generated documentation: {file_path}")
+
+if __name__ == "__main__":
+    generate_docs()