proxy using msgraph sdk

Author: geekwhocodes

src/source_msgraph/graph.py

@@ -8,11 +8,13 @@
 from kiota_abstractions.base_request_configuration import RequestConfiguration
 from msgraph.generated.sites.item.lists.item.items.items_request_builder import ItemsRequestBuilder
 import json
-from kiota_serialization_json.json_serialization_writer_factory import JsonSerializationWriterFactory
+
 from msgraph import GraphServiceClient
 from azure.identity import ClientSecretCredential
 
 from source_msgraph.async_interator import AsyncToSyncIterator
+from source_msgraph.msgraph_spark.fetcher import MicrosoftGraphFetcher
+from source_msgraph.msgraph_spark.options import MicrosoftGraphOptions
 
 tenant_id = "7c78a7a0-8b3b-4d54-8c3c-ca6ab4da029f"
 client_id = "59c8283d-4a90-44a9-9a58-579a0e511168"
@@ -24,17 +26,11 @@
 
 # from microsoft.kiota.serialization.json.json_serialization_writer_factory import JsonSerializationWriterFactory
 
-# Convert to JSON using Kiota
-writer_factory = JsonSerializationWriterFactory()
-writer = writer_factory.get_serialization_writer("application/json")
+
 graph_client = GraphServiceClient(credentials=credentials, scopes=[
                                   'https://graph.microsoft.com/.default'])
 
 
-def to_json(value):
-    value.serialize(writer)
-    # Get JSON string
-    return json.loads((writer.get_serialized_content().decode("utf-8")))
 
 
 async def fetch_items_async(graph_client, site_id, list_id, **params):
@@ -57,4 +53,32 @@ async def fetch_items_async(graph_client, site_id, list_id, **params):
 
 def fetch_items_sync(graph_client, site_id, list_id, params):
     async_gen = fetch_items_async(graph_client, site_id, list_id, **params)
-    return AsyncToSyncIterator(async_gen)
+    return AsyncToSyncIterator(async_gen)
+
+
+def main():
+    options = MicrosoftGraphOptions(
+        tenant_id=tenant_id,
+        client_id=client_id,
+        client_secret=client_secret,
+        resource="sites/by_site_id/lists/by_list_id/items",
+        params={
+            "top": 1,
+            "expand": ["fields"]
+        }
+    )
+    resource_path = "sites/by_site_id/lists/by_list_id/items"
+    params = {
+        "site_id": "37d7dde8-0b6b-4b7c-a2fd-2e217f54a263",  # Actual site ID
+        "list_id": "5ecf26db-0161-4069-b763-856217415099",  # Actual list ID
+        "top": 5,
+        "expand": ["fields"]
+    }
+    fetcher = MicrosoftGraphFetcher(graph_client, resource_path, params)
+    async_gen = fetcher.fetch_data()
+    for row in AsyncToSyncIterator(async_gen):
+        print(row)
+
+
+if __name__ == "__main__":
+    main()

src/source_msgraph/msgraph_spark/client.py

@@ -0,0 +1,23 @@
+from typing import Union
+from pyspark.sql.types import StructType
+
+from source_msgraph.async_interator import AsyncToSyncIterator
+from source_msgraph.msgraph_spark.fetcher import MicrosoftGraphFetcher
+from source_msgraph.msgraph_spark.options import MicrosoftGraphOptions
+from azure.identity import ClientSecretCredential
+from msgraph import GraphServiceClient
+
+def iter_records(schema: Union[StructType, str], options: MicrosoftGraphOptions):
+    # Authenticate using ClientSecretCredential
+    credentials = ClientSecretCredential(
+        tenant_id=options.tenant_id,
+        client_id=options.client_id,
+        client_secret=options.client_secret
+    )
+
+    # Initialize Graph Client
+    client = GraphServiceClient(credentials=credentials, scopes=["https://graph.microsoft.com/.default"])
+
+    fetcher = MicrosoftGraphFetcher(client, options.resource_path, options.query_params, options.resource_options)
+    async_gen = fetcher.fetch_data()
+    return AsyncToSyncIterator(async_gen)

src/source_msgraph/msgraph_spark/fetcher.py

@@ -0,0 +1,139 @@
+import importlib
+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 typing import Dict, Any
+import asyncio
+
+class MicrosoftGraphFetcher:
+    def __init__(self, graph_client: GraphServiceClient, resource_path: str, query_params: Dict[str, Any], resource_params: Dict[str, str]):
+        """
+        Initializes the fetcher with the Graph client, resource path, and query parameters.
+
+        :param graph_client: The authenticated GraphServiceClient instance.
+        :param resource_path: The resource path (e.g., "sites/by_site_id/lists/by_list_id/items").
+        :param params: Query parameters for the request.
+        """
+        self.graph_client = graph_client
+        self.resource_path = resource_path
+        self.query_params = query_params
+        self.resource_params = resource_params
+
+    def _get_request_builder_class(self):
+        """
+        Dynamically resolves the correct RequestBuilder class from `msgraph.generated`.
+
+        Example: For "sites/by_site_id/lists/by_list_id/items", it will resolve:
+            `msgraph.generated.sites.item.lists.item.items.items_request_builder.ItemsRequestBuilder`
+        """
+        path_parts = self.resource_path.split("/")
+        base_module = "msgraph.generated"
+        module_path = []
+        class_name = "RequestBuilder"  # Default fallback
+
+        i = 0
+        while i < len(path_parts):
+            part = path_parts[i]
+
+            if part.startswith("by_"):  # Handling {site_id}, {list_id}, etc.
+                module_path.append("item")  # `by_site_id` means `.item`
+                
+            else:
+                module_path.append(part)
+
+            i += 1
+
+        # Construct the full module path
+        module_name = f"{base_module}." + ".".join(module_path) + f".{module_path[-1]}_request_builder"
+
+        try:
+            module = importlib.import_module(module_name)
+            pascal_case_class = self._pascal_case(f"{module_path[-1]}_request_builder")
+
+            for attr in dir(module):
+                if attr == pascal_case_class:
+                    return getattr(module, attr)
+
+            raise ValueError(f"Could not find {pascal_case_class} in {module_name}")
+
+        except ModuleNotFoundError:
+            raise ValueError(f"Could not resolve RequestBuilder for resource path: {self.resource_path}")
+
+    def _pascal_case(self, snake_str: str) -> str:
+        """
+        Converts snake_case to PascalCase.
+        Example: "items_request_builder" -> "ItemsRequestBuilder"
+        """
+        return "".join(word.title() for word in snake_str.split("_"))
+    
+    def _get_query_parameters_class(self, request_builder_class):
+        """
+        Fetches the corresponding `RequestBuilderGetQueryParameters` class dynamically.
+        """
+        for attr in dir(request_builder_class):
+            if attr.endswith("RequestBuilderGetQueryParameters"):
+                return getattr(request_builder_class, attr)
+        raise ValueError(f"No QueryParameters class found for {request_builder_class.__name__}")
+
+    async def fetch_data(self):
+        """
+        Fetches data from Microsoft Graph using the dynamically built request.
+        Handles pagination automatically.
+        """
+        request_builder_class = self._get_request_builder_class()
+        query_params_class = self._get_query_parameters_class(request_builder_class)
+
+        
+        # Create Query Parameters object
+        valid_params = {p for p in query_params_class.__annotations__.keys()}
+        filtered_params = {k: v for k, v in self.query_params.items() if k in valid_params}
+        query_parameters = query_params_class(**filtered_params)
+        request_configuration = RequestConfiguration(
+            query_parameters=query_parameters,
+        )
+        
+        # Get Request Builder Instance from Graph Client
+        builder = self._get_request_builder_instance(request_builder_class)
+
+        try:
+            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 _get_request_builder_instance(self, request_builder_class):
+        """
+        Uses the `graph_client` to resolve the correct instance of the request builder dynamically,
+        passing actual `site_id`, `list_id`, etc., instead of placeholders.
+        """
+        parts = self.resource_path.split("/")
+        builder = self.graph_client
+
+        i = 0
+        while i < len(parts):
+            part = parts[i]
+
+            if part.startswith("by_"):  # Handling "by_site_id", "by_list_id", etc.
+                param_name = part[3:]  # Extract parameter name (e.g., "site_id", "list_id")
+                actual_id = self.resource_params.get(param_name)  # Get actual ID from user input
+                if not actual_id:
+                    raise ValueError(f"Missing required parameter: {param_name}")
+                method_name = part  # Keep "by_site_id" format
+
+                if hasattr(builder, method_name):
+                    builder = getattr(builder, method_name)(actual_id)  # Pass actual ID
+            elif hasattr(builder, part):
+                builder = getattr(builder, part)
+            else:
+                raise ValueError(f"Invalid resource path: '{part}' not found in {builder}")
+
+            i += 1
+
+        return builder

src/source_msgraph/msgraph_spark/options.py

@@ -0,0 +1,55 @@
+from dataclasses import dataclass, field
+from typing import Optional, Dict, Any
+
+
+@dataclass
+class MicrosoftGraphOptions:
+    """Options for Microsoft Graph API requests with strict resource_path validation."""
+    tenant_id: str
+    client_id: str
+    client_secret: str
+    resource_path: str  # Unique identifier, e.g., "sites/{site_id}/lists/{list_id}"
+    query_params: Dict[str, Any] = field(default_factory=dict)  # Query params like 'top', 'filter'
+    resource_options: Dict[str, Any] = field(default_factory=dict)  # Other resource-specific options
+
+    def __post_init__(self):
+        """Validate query params and resource options based on full resource_path."""
+        self.validate_query_params()
+        self.validate_resource_options()
+        
+
+    def validate_query_params(self):
+        """Validate query parameters against the allowed list for the full resource path."""
+        valid_query_options = self.get_valid_query_options()
+        invalid_keys = [key for key in self.query_params if key not in valid_query_options]
+        if invalid_keys:
+            raise ValueError(f"Invalid query parameters for {self.resource_path}: {invalid_keys}")
+
+    def validate_resource_options(self):
+        """Validate resource-specific options based on the full resource_path."""
+        valid_options = self.get_valid_resource_options()
+        invalid_keys = [key for key in self.resource_options if key not in valid_options]
+        if invalid_keys:
+            raise ValueError(f"Invalid resource options for {self.resource_path}: {invalid_keys}")
+
+    def get_valid_query_options(self):
+        """Returns allowed query parameters based on the full resource path."""
+        valid_query_map = {
+            "sites/by_site_id/lists/by_list_id/items": {"top", "filter", "orderby", "expand"},
+        }
+        return valid_query_map.get(self.resource_path, set())
+
+    def get_valid_resource_options(self):
+        """Returns allowed resource options based on the full resource path."""
+        valid_resource_map = {
+            "sites/by_site_id/lists/by_list_id/items": {"site_id", "list_id"},
+           
+        }
+        return valid_resource_map.get(self.resource_path, set())
+
+    def get_request_parameters(self):
+        """Returns structured parameters for Microsoft Graph SDK."""
+        return {
+            "query_params": self.query_params,
+            "resource_options": self.resource_options
+        }

src/source_msgraph/msgraph_spark/utils.py

@@ -0,0 +1,15 @@
+from kiota_serialization_json.json_serialization_writer_factory import JsonSerializationWriterFactory
+import json
+# Convert to JSON using Kiota
+writer_factory = JsonSerializationWriterFactory()
+writer = writer_factory.get_serialization_writer("application/json")
+
+def to_json(value):
+    value.serialize(writer)
+    # Get JSON string
+    return json.loads((writer.get_serialized_content().decode("utf-8")))
+
+def to_jsonValue(value):
+    value.serialize(writer)
+    # Get JSON string
+    return str(json.loads((writer.get_serialized_content().decode("utf-8"))))

src/source_msgraph/resource_resolver.py

@@ -0,0 +1,18 @@
+import importlib
+
+def resolve_resource_class(resource_path: str):
+    """Resolve the appropriate resource handler based on the resource path."""
+    parts = resource_path.split('/')
+    module_name = "source_msgraph"
+
+    for depth in range(min(len(parts), 5)):  # Limit depth to 5
+        module_name += f".{parts[depth]}"
+        try:
+            module = importlib.import_module(module_name)
+            if hasattr(module, "ResourceHandler"):
+                return module.ResourceHandler
+        except ModuleNotFoundError:
+            continue  # Try next level
+    
+    # TODO: Implement BaseResource if required at all
+    raise ValueError(f"No handler found for resource: {resource_path}")

src/source_msgraph/resources/base_options.py

@@ -0,0 +1,18 @@
+from dataclasses import dataclass, field
+from typing import Optional, Dict, Any
+
+
+@dataclass
+class MicrosoftGraphResourceOptions:
+    """Base dataclass for Microsoft Graph resource options."""
+    
+    # resource: str  # e.g., "users", "sites", "groups"
+    # sub_resource: Optional[str] = None  # e.g., for sites: "lists", "drive"
+    # params: Dict[str, Any] = field(default_factory=dict)  # Additional query params
+    
+    # def validate(self):
+    #     """Base validation for all resources."""
+    #     if not self.resource:
+    #         raise ValueError("Resource is required.")
+    #     if self.sub_resource and not isinstance(self.sub_resource, str):
+    #         raise ValueError("Sub-resource must be a string.")

src/source_msgraph/resources/sites/lists.py

@@ -0,0 +1,48 @@
+from dataclasses import dataclass
+from typing import Optional
+from msgraph import GraphServiceClient
+from msgraph.generated.sites.item.lists.item.items.items_request_builder import ItemsRequestBuilder
+from kiota_abstractions.base_request_configuration import RequestConfiguration
+from source_msgraph.async_interator import AsyncToSyncIterator
+from source_msgraph.resources.base_options import MicrosoftGraphResourceOptions
+
+@dataclass
+class SharePointListOptions(MicrosoftGraphResourceOptions):
+    site_id: str
+    list_id: Optional[str] = None
+
+    def validate(self):
+        super().validate()
+        if not self.site_id:
+            raise ValueError("site_id is required for SharePoint lists.")
+        if self.sub_resource == "lists" and not self.list_id:
+            raise ValueError("list_id is required when accessing lists.")
+
+
+class ResourceHandler:
+    def __init__(self, client: GraphServiceClient, params: dict):
+        self.client = client
+        self.params = SharePointListOptions(**params)
+
+    async def fetch_items_async(self, site_id, list_id, **params):
+        query_parameters = ItemsRequestBuilder.ItemsRequestBuilderGetQueryParameters(**params,
+            expand=["fields"])
+        request_configuration = RequestConfiguration(
+            query_parameters=query_parameters,
+        )
+
+        items = await self.client.sites.by_site_id(site_id).lists.by_list_id(list_id).items.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 self.client.sites.by_site_id(site_id).lists.by_list_id(list_id).items.with_url(items.odata_next_link).get()
+
+
+    def fetch_items_sync(self, site_id, list_id, params):
+        async_gen = self.fetch_items_async(self.client, site_id, list_id, **params)
+        return AsyncToSyncIterator(async_gen)
+