🔍 Code Extractor

GraphClient is a client class for interacting with Microsoft Graph API, providing authentication and access to various Microsoft 365 services including users, groups, drives, teams, and more.

Purpose

GraphClient serves as the main entry point for accessing Microsoft Graph API resources. It handles authentication via multiple methods (certificate, client secret, username/password, interactive), manages HTTP requests with proper authentication headers, and provides property-based access to all major Microsoft 365 services like Azure AD, OneDrive, SharePoint, Teams, Intune, and more. It extends ClientRuntimeContext to provide OData query capabilities and batch request support.

Source Code

class GraphClient(ClientRuntimeContext):
    """Graph Service client"""

    def __init__(self, acquire_token_callback):
        # type: (Callable[[], dict]) -> None
        """
        :param () -> dict acquire_token_callback: Acquire token function
        """
        super(GraphClient, self).__init__()
        self._pending_request = None
        self._resource = "https://graph.microsoft.com"
        self._authority_host_url = "https://login.microsoftonline.com"
        self._acquire_token_callback = acquire_token_callback

    @staticmethod
    def with_certificate(
        tenant, client_id, thumbprint, private_key, scopes=None, token_cache=None
    ):
        """
        Initializes the confidential client with client certificate

        :param str tenant: Tenant name, for example: contoso.onmicrosoft.com
        :param str client_id: The OAuth client id of the calling application.
        :param str thumbprint: Thumbprint
        :param str private_key: Private key
        :param list[str] or None scopes: Scopes requested to access an API
        :param Any token_cache: Default cache is in memory only,
        Refer https://msal-python.readthedocs.io/en/latest/#msal.SerializableTokenCache
        """
        if scopes is None:
            scopes = ["https://graph.microsoft.com/.default"]
        authority_url = "https://login.microsoftonline.com/{0}".format(tenant)
        import msal

        app = msal.ConfidentialClientApplication(
            client_id,
            authority=authority_url,
            client_credential={
                "thumbprint": thumbprint,
                "private_key": private_key,
            },
            token_cache=token_cache  # Default cache is in memory only.
            # You can learn how to use SerializableTokenCache from
            # https://msal-python.readthedocs.io/en/latest/#msal.SerializableTokenCache
        )

        def _acquire_token():
            return app.acquire_token_for_client(scopes=scopes)

        return GraphClient(_acquire_token)

    @staticmethod
    def with_client_secret(
        tenant, client_id, client_secret, scopes=None, token_cache=None
    ):
        # type: (str, str, str, List[str], Any) -> "GraphClient"
        """
        Initializes the confidential client with client secret

        :param str tenant: Tenant name, for example: contoso.onmicrosoft.com
        :param str client_id: The OAuth client id of the calling application.
        :param str client_secret: Client secret
        :param list[str] or None scopes: Scopes requested to access an API
        :param Any token_cache: Default cache is in memory only,
        Refer https://msal-python.readthedocs.io/en/latest/#msal.SerializableTokenCache
        """
        if scopes is None:
            scopes = ["https://graph.microsoft.com/.default"]
        authority_url = "https://login.microsoftonline.com/{0}".format(tenant)
        import msal

        app = msal.ConfidentialClientApplication(
            client_id,
            authority=authority_url,
            client_credential=client_secret,
            token_cache=token_cache,
        )

        def _acquire_token():
            return app.acquire_token_for_client(scopes=scopes)

        return GraphClient(_acquire_token)

    @staticmethod
    def with_token_interactive(tenant, client_id, username=None, scopes=None):
        # type: (str, str, Optional[str], List[str]) -> "GraphClient"
        """
        Initializes the client via user credentials
        Note: only works if your app is registered with redirect_uri as http://localhost

        :param str tenant: Tenant name, for example: contoso.onmicrosoft.com
        :param str client_id: The OAuth client id of the calling application.
        :param str username: Typically a UPN in the form of an email address.
        :param list[str] or None scopes: Scopes requested to access an API
        """
        if scopes is None:
            scopes = ["https://graph.microsoft.com/.default"]
        authority_url = "https://login.microsoftonline.com/{0}".format(tenant)
        import msal

        app = msal.PublicClientApplication(client_id, authority=authority_url)

        def _acquire_token():
            # The pattern to acquire a token looks like this.
            result = None

            # Firstly, check the cache to see if this end user has signed in before
            accounts = app.get_accounts(username=username)
            if accounts:
                chosen = accounts[0]  # Assuming the end user chose this one to proceed
                # Now let's try to find a token in cache for this account
                result = app.acquire_token_silent(scopes, account=chosen)

            if not result:
                result = app.acquire_token_interactive(
                    scopes,
                    login_hint=username,
                )
            return result

        return GraphClient(_acquire_token)

    @staticmethod
    def with_username_and_password(tenant, client_id, username, password, scopes=None):
        # type: (str, str, str, str, List[str]) -> "GraphClient"
        """
        Initializes the client via user credentials

        :param str tenant: Tenant name, for example: contoso.onmicrosoft.com
        :param str client_id: The OAuth client id of the calling application.
        :param str username: Typically a UPN in the form of an email address.
        :param str password: The password.
        :param list[str] or None scopes: Scopes requested to access an API
        """
        if scopes is None:
            scopes = ["https://graph.microsoft.com/.default"]
        authority_url = "https://login.microsoftonline.com/{0}".format(tenant)
        import msal

        app = msal.PublicClientApplication(
            authority=authority_url,
            client_id=client_id,
        )

        def _acquire_token():
            result = None
            accounts = app.get_accounts(username=username)
            if accounts:
                result = app.acquire_token_silent(scopes, account=accounts[0])

            if not result:
                result = app.acquire_token_by_username_password(
                    username=username,
                    password=password,
                    scopes=scopes,
                )
            return result

        return GraphClient(_acquire_token)

    def execute_batch(self, items_per_batch=100):
        """Constructs and submit a batch request

        :param int items_per_batch: Maximum to be selected for bulk operation
        """
        batch_request = ODataV4BatchRequest(V4JsonFormat())
        batch_request.beforeExecute += self._authenticate_request
        while self.has_pending_request:
            qry = self._get_next_query(items_per_batch)
            batch_request.execute_query(qry)
        return self

    def pending_request(self):
        # type: () -> ODataRequest
        if self._pending_request is None:
            self._pending_request = ODataRequest(V4JsonFormat())
            self._pending_request.beforeExecute += self._authenticate_request
            self._pending_request.beforeExecute += self._build_specific_query
        return self._pending_request

    def service_root_url(self):
        # type: () -> str
        return "https://graph.microsoft.com/v1.0"

    def _build_specific_query(self, request):
        # type: (RequestOptions) -> None
        """Builds Graph specific HTTP request."""
        if isinstance(self.current_query, UpdateEntityQuery):
            request.method = HttpMethod.Patch
        elif isinstance(self.current_query, DeleteEntityQuery):
            request.method = HttpMethod.Delete

    def _authenticate_request(self, request):
        # type: (RequestOptions) -> None
        """Authenticate request."""
        token_json = self._acquire_token_callback()
        token = TokenResponse.from_json(token_json)
        request.ensure_header("Authorization", "Bearer {0}".format(token.accessToken))

    @property
    def admin(self):
        """A container for administrator functionality for SharePoint and OneDrive."""
        return Admin(self, ResourcePath("admin"))

    @property
    def app_catalogs(self):
        """A container for apps from the Microsoft Teams app catalog."""
        return AppCatalogs(self, ResourcePath("appCatalogs"))

    @property
    def me(self):
        """The Me endpoint is provided as a shortcut for specifying the current user"""
        return User(self, MePath())

    @property
    def device_management(self):
        """Singleton entity that acts as a container for all device management functionality."""
        return DeviceManagement(self, ResourcePath("deviceManagement"))

    @property
    def device_app_management(self):
        """Singleton entity that acts as a container for all device and app management functionality."""
        return DeviceAppManagement(self, ResourcePath("deviceAppManagement"))

    @property
    def drives(self):
        """Get one drives"""
        return EntityCollection(self, Drive, ResourcePath("drives"))

    @property
    def users(self):
        """Users container"""
        return UserCollection(self, ResourcePath("users"))

    @property
    def domains(self):
        """Alias to domains"""
        return EntityCollection(self, Domain, ResourcePath("domains"))

    @property
    def groups(self):
        """Get groups"""
        return GroupCollection(self, ResourcePath("groups"))

    @property
    def invitations(self):
        """Get invitations"""
        return InvitationCollection(self, ResourcePath("invitations"))

    @property
    def identity_protection(self):
        return IdentityProtectionRoot(self, ResourcePath("identityProtection"))

    @property
    def sites(self):
        """Sites container"""
        return SitesWithRoot(self, ResourcePath("sites"))

    @property
    def shares(self):
        """Shares container"""
        return SharesCollection(self, ResourcePath("shares"))

    @property
    def directory_objects(self):
        """Directory Objects container"""
        return DirectoryObjectCollection(self, ResourcePath("directoryObjects"))

    @property
    def teams(self):
        """Teams container"""
        return TeamCollection(self, ResourcePath("teams"))

    @property
    def chats(self):
        """Chats container"""
        return ChatCollection(self, ResourcePath("chats"))

    @property
    def group_setting_templates(self):
        """Group setting templates represent system-defined settings available to the tenant."""
        return EntityCollection(
            self, GroupSettingTemplate, ResourcePath("groupSettingTemplates")
        )

    @property
    def contacts(self):
        """Get the list of organizational contacts for this organization."""
        return DeltaCollection(self, OrgContact, ResourcePath("contacts"))

    @property
    def directory(self):
        """Represents a deleted item in the directory"""
        return Directory(self, ResourcePath("directory"))

    @property
    def directory_roles(self):
        """Represents a directory roles in the directory"""
        return DeltaCollection(self, DirectoryRole, ResourcePath("directoryRoles"))

    @property
    def directory_role_templates(self):
        """Represents a directory role templates in the directory"""
        from office365.directory.rolemanagement.template import DirectoryRoleTemplate

        return EntityCollection(
            self, DirectoryRoleTemplate, ResourcePath("directoryRoleTemplates")
        )

    @property
    def identity_providers(self):
        """"""
        return EntityCollection(
            self, IdentityProvider, ResourcePath("identityProviders")
        )

    @property
    def identity(self):
        return IdentityContainer(self, ResourcePath("identity"))

    @property
    def application_templates(self):
        """Get the list of application templates in this organization."""
        return EntityCollection(
            self, ApplicationTemplate, ResourcePath("applicationTemplates")
        )

    @property
    def applications(self):
        """Get the list of applications in this organization."""
        return ApplicationCollection(self, ResourcePath("applications"))

    @property
    def service_principals(self):
        """Retrieve a list of servicePrincipal objects."""
        return ServicePrincipalCollection(self, ResourcePath("servicePrincipals"))

    @property
    def organization(self):
        """"""
        return EntityCollection(self, Organization, ResourcePath("organization"))

    @property
    def subscribed_skus(self):
        """Get the list of commercial subscriptions that an organization has acquired"""
        return EntityCollection(self, SubscribedSku, ResourcePath("subscribedSkus"))

    @property
    def group_lifecycle_policies(self):
        """A collection of lifecycle policies for a Microsoft 365 groups."""
        return EntityCollection(
            self, GroupLifecyclePolicy, ResourcePath("groupLifecyclePolicies")
        )

    @property
    def group_settings(self):
        """Represents a directory roles in the directory"""
        from office365.directory.groups.setting import GroupSetting

        return GroupSetting(self, ResourcePath("groupSettings"))

    @property
    def communications(self):
        """
        Cloud communications API endpoint
        """
        return CloudCommunications(self, ResourcePath("communications"))

    @property
    def identity_governance(self):
        """
        The identity governance singleton
        """
        return IdentityGovernance(self, ResourcePath("identityGovernance"))

    @property
    def information_protection(self):
        """
        Exposes methods that you can use to get Microsoft Purview Information Protection labels and label policies.
        """
        return InformationProtection(self, ResourcePath("informationProtection"))

    @property
    def subscriptions(self):
        """
        Retrieve the properties and relationships of webhook subscriptions,
        based on the app ID, the user, and the user's role with a tenant.
        """
        return SubscriptionCollection(self, ResourcePath("subscriptions"))

    @property
    def connections(self):
        """Get a list of the externalConnection objects and their properties."""
        from office365.search.external.connection import ExternalConnection

        return EntityCollection(self, ExternalConnection, ResourcePath("connections"))

    @property
    def tenant_relationships(self):
        """Represent the various type of tenant relationships."""
        return TenantRelationship(self, ResourcePath("tenantRelationships"))

    @property
    def audit_logs(self):
        """Gets the list of audit logs generated by Azure Active Directory."""
        return AuditLogRoot(self, ResourcePath("auditLogs"))

    @property
    def places(self):
        """Gets all places in a tenant"""
        return EntityCollection(self, Place, ResourcePath("places"))

    @property
    def room_lists(self):
        """Gets all the room lists in a tenant"""
        return EntityCollection(
            self,
            RoomList,
            ResourcePath("microsoft.graph.roomlist", ResourcePath("places")),
        )

    @property
    def reports(self):
        """Represents a container for Azure Active Directory (Azure AD) reporting resources."""
        return ReportRoot(self, ResourcePath("reports"))

    @property
    def role_management(self):
        """Represents a Microsoft 365 role-based access control (RBAC) role management container"""
        return RoleManagement(self, ResourcePath("roleManagement"))

    @property
    def solutions(self):
        return SolutionsRoot(self, ResourcePath("solutions"))

    @property
    def teams_templates(self):
        """
        Get the list of teams templates.
        """
        return EntityCollection(self, TeamsTemplate, ResourcePath("teamsTemplates"))

    @property
    def planner(self):
        """
        The planner resource is the entry point for the Planner object model.
        It returns a singleton planner resource. It doesn't contain any usable properties.
        """
        return Planner(self, ResourcePath("planner"))

    @property
    def permission_grants(self):
        """
        List all resource-specific permission grants
        """
        return EntityCollection(
            self, ResourceSpecificPermissionGrant, ResourcePath("permissionGrants")
        )

    @property
    def print(self):
        """
        Used to manage printers and print jobs within Universal Print.
        """
        from office365.intune.printing.print import Print

        return Print(self, ResourcePath("print"))

    @property
    def search(self):
        """
        The search endpoint is the entry point for Microsoft Search API to query data.
        """
        return SearchEntity(self, ResourcePath("search"))

    @property
    def employee_experience(self):
        """
        An alias to EmployeeExperience
        """
        return EmployeeExperience(self, ResourcePath("employeeExperience"))

    @property
    def education(self):
        """
        The /education namespace exposes functionality that is specific to the education sector.
        """
        return EducationRoot(self, ResourcePath("education"))

    @property
    def policies(self):
        """Resource type exposing navigation properties for the policies singleton."""
        return PolicyRoot(self, ResourcePath("policies"))

    @property
    def external(self):
        """A logical container for external sources."""
        return External(self, ResourcePath("external"))

    @property
    def security(self):
        """The security resource is the entry point for the Security object model.
        It returns a singleton security resource. It doesn't contain any usable properties.
        """
        return Security(self, ResourcePath("security"))

    @property
    def schema_extensions(self):
        """Get a list of schemaExtension objects in your tenant"""
        return EntityCollection(self, SchemaExtension, ResourcePath("schemaExtensions"))

Parameters

Parameter Details

acquire_token_callback: A callable function that returns a dictionary containing authentication token information. This function is invoked before each API request to obtain a valid access token. The returned dictionary should contain token details compatible with TokenResponse.from_json().

Return Value

The constructor returns a GraphClient instance configured with the provided token acquisition callback. Static factory methods (with_certificate, with_client_secret, with_username_and_password, with_token_interactive) return fully configured GraphClient instances. The execute_batch method returns self for method chaining. Property accessors return various collection and entity objects representing Microsoft Graph resources.

Class Interface

Methods

Attributes

Dependencies

• msal
• office365

Required Imports

from office365.graph_client import GraphClient

Conditional/Optional Imports

These imports are only needed under specific conditions:

import msal

Usage Example

# Using client secret authentication
client = GraphClient.with_client_secret(
    tenant='contoso.onmicrosoft.com',
    client_id='your-client-id',
    client_secret='your-client-secret'
)

# Access users
users = client.users.get().execute_query()
for user in users:
    print(user.display_name)

# Access current user
me = client.me.get().execute_query()
print(me.mail)

# Access groups
groups = client.groups.get().execute_query()

# Using certificate authentication
client_cert = GraphClient.with_certificate(
    tenant='contoso.onmicrosoft.com',
    client_id='your-client-id',
    thumbprint='cert-thumbprint',
    private_key='-----BEGIN PRIVATE KEY-----\n...'
)

# Using username/password
client_user = GraphClient.with_username_and_password(
    tenant='contoso.onmicrosoft.com',
    client_id='your-client-id',
    username='user@contoso.onmicrosoft.com',
    password='password'
)

# Batch operations
client.execute_batch(items_per_batch=50)

Best Practices

• Always use one of the static factory methods (with_certificate, with_client_secret, etc.) to instantiate GraphClient rather than calling __init__ directly
• Store sensitive credentials (client secrets, passwords, private keys) securely using environment variables or secure vaults, never hardcode them
• Use certificate-based authentication for production applications as it's more secure than client secrets
• Implement token caching using msal.SerializableTokenCache to avoid unnecessary token acquisitions and improve performance
• Use appropriate scopes for your application's needs; the default '.default' scope grants all permissions configured in Azure AD
• Call execute_query() on collections and entities to actually execute the API request and retrieve data
• Use execute_batch() for bulk operations to reduce API calls and improve performance
• Handle token expiration gracefully; the client automatically acquires new tokens via the callback
• Be aware that property accessors (like .users, .groups) return collection/entity objects that require execute_query() to fetch data
• The client maintains state through _pending_request; avoid sharing client instances across threads without proper synchronization
• For interactive authentication (with_token_interactive), ensure your app is registered with redirect URI as http://localhost
• Username/password authentication may not work with accounts that have MFA enabled; prefer interactive or certificate-based auth

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class SharePointGraphClient 71.3% similar

  SharePoint client using Microsoft Graph API. This bypasses SharePoint REST API app-only token issues.

  From: /tf/active/vicechatdev/SPFCsync/sharepoint_graph_client.py

• class GraphRequest 67.5% similar

  GraphRequest is a specialized OData request class configured for Microsoft Graph API interactions using V4 JSON format.

  From: /tf/active/vicechatdev/SPFCsync/venv/lib64/python3.11/site-packages/office365/graph_request.py

• function test_graph_client 66.2% similar

  A test function that validates the SharePoint Graph API client by testing authentication, document listing, and file download capabilities.

  From: /tf/active/vicechatdev/SPFCsync/test_graph_client.py

• class Group 66.1% similar

  Represents an Azure Active Directory (Azure AD) group, which can be an Office 365 group or a security group, providing methods to manage group operations, memberships, and associated resources.

  From: /tf/active/vicechatdev/SPFCsync/venv/lib64/python3.11/site-packages/office365/directory/groups/group.py

• class GroupCollection 60.3% similar

  A collection class for managing Microsoft Graph API Group resources, providing methods to create, retrieve, and manage groups including Microsoft 365 groups and Security groups.

  From: /tf/active/vicechatdev/SPFCsync/venv/lib64/python3.11/site-packages/office365/directory/groups/collection.py