Integrating Google Workspace: A Simplified Guide

Admin SDK API

Recently, I faced the challenge of integrating Google Workspace for a project aimed at pulling data from Chrome OS devices. The setup process was initially daunting, especially due to the lack of straightforward documentation. This blog aims to demystify the process, particularly for those looking to use Google Admin Console APIs without constant user input, ideal for automated tasks like cron jobs.

Part 1: Setting Up Authentication

Step 1: Create a Google Workspace Account

Before you begin, ensure you have a Google Workspace account which will be central to this integration.

Step 2: Set Up a Google Cloud Account

Similarly, you will need a Google Cloud account. These two platforms will link together to manage your API access.

Step 3: Create a Service Account

1. Log in to your Google Cloud Account.
2. Navigate to IAM & Admin > Service Accounts.

1. Or you can go directly to https://console.developers.google.com/iam-admin/serviceaccounts
2. Create a new service account and name it appropriately for your project.
3. Select your new service account and go to the Keys tab.
4. In the Add key drop-down, choose Create new key.
5. Download the JSON key file — this file is crucial as it contains your credentials. Handle it with care and ensure it’s securely stored.

For more details on this process, refer to Google’s official documentation on Using OAuth 2.0 for Server to Server Applications.

Part 2: Enable API Access in Google Admin

Step 1: Enable the Google Admin API

1. Use this link to enable the Google Admin API: Enable API.

Step 2: Manage API Controls

1. Navigate to the Google Admin Console.
2. Click on API Controls in the sidebar.
3. Go to Domain-wide Delegation and click on Manage Domain-wide Delegation.

4. Click Add New and enter the Unique ID/Client ID from your service account.

Step 3: Set API Scopes

Decide on the necessary scopes for your project. For mine, I added the following scopes:

• [https://www.googleapis.com/auth/admin.directory.user.readonly](https://www.googleapis.com/auth/admin.directory.user.readonly)
• [https://www.googleapis.com/auth/admin.directory.device.chromeos](https://www.googleapis.com/auth/admin.directory.device.chromeos)
• [https://www.googleapis.com/auth/admin.directory.orgunit.readonly](https://www.googleapis.com/auth/admin.directory.orgunit.readonly)

For a complete list of available scopes, check the Admin SDK Directory API documentation.

Connecting to the Google Admin SDK with Python

In the previous section, we discussed setting up authentication for Google Workspace. Now, let’s explore how to utilize the Google Admin SDK APIs using Python to efficiently manage Chrome OS devices.

Installation of Necessary Packages

Before we begin coding, make sure you have the required Python libraries installed. Run the following commands in your terminal to install them:

pip install google-api-python-client
pip install google-auth-httplib2
pip install google-auth-oauthlib

Configuring Authentication Headers

To interact with the Google APIs, you need properly configured headers. Below, we use asynchronous Python code to generate these headers using a service account. Note that the JSON key obtained previously during authentication setup is required here. The function get_secret_value is designed to securely retrieve the key as a string. After fetching, I then convert this string into a dictionary for use in our authentication process. The delegated_email parameter represents an admin user within the Google Workspace account.

import asyncio
import json
import os
import sys

import google.auth.transport.requests
from google.oauth2 import service_account

async def get_google_admin_api_headers(*, delegated_email: str) -> None | dict:
    """Gets the headers for the Google Admin API."""
    scopes = [
        "https://www.googleapis.com/auth/admin.directory.user.readonly",
        "https://www.googleapis.com/auth/admin.directory.device.chromeos",
        "https://www.googleapis.com/auth/admin.directory.orgunit.readonly",
    ]
    try:
        service_account_key_string = await get_secret_value(
            secret_name="google-admin-json"
        )
        service_account_key: dict = json.loads(service_account_key_string)
        credentials = service_account.Credentials.from_service_account_info(
            service_account_key, scopes=scopes
        ).with_subject(delegated_email)
        request = google.auth.transport.requests.Request()
        credentials.refresh(request)
        return {"Authorization": f"Bearer {credentials.token}"}
    except Exception as e:
        logger.error(f"Error getting Google Admin API headers: {e}")
        return None

Fetching All Chrome OS Devices

Once the authentication headers are set, you can fetch all Chrome OS devices linked to your account. The following function utilizes the HTTP client library httpx to make asynchronous API calls.

The tenant_id is your Customer ID.

Find your customer ID — Google Workspace Admin Help

async def get_all_chrome_os_devices(
    *, google_api_headers: dict, tenant_id: str, one_page: bool = False
) -> list[ChromeOSDevice] | None:
    """Gets all Chrome OS device from the Google Admin SDK."""
    url = f"https://admin.googleapis.com/admin/directory/v1/customer/{tenant_id}/devices/chromeos"
    all_devices_data = []

    async with httpx.AsyncClient(headers=google_api_headers, timeout=120) as client:
        while url:
            try:
                response = await client.get(url)
                response.raise_for_status()  # Raises an exception for 400 and 500 status codes
                data_json = response.json()

            except httpx.HTTPStatusError as e:
                if response.status_code == 504:
                    logger.info("Request timed out.")
                    return None
                logger.error(
                    f"Error fetching Chrome OS devices.\nHttpError: {e}\nStatus code: {response.status_code}\nResponse: {response.text}"
                )
                return None

            devices = data_json.get("chromeosdevices", [])
            all_devices_data.extend(devices)
            if one_page is True:
                break
            # Google uses nextPageToken for pagination
            page_token = data_json.get("nextPageToken")
            url = (
                f"{url}&pageToken={page_token}" if page_token else None
            )  # Break the loop if there's no nextPageToken

    return [ChromeOSDevice(**device) for device in all_devices_data]

Defining Our Pydantic Model for Chrome OS Devices

We have created a Pydantic model to represent the details of a Chrome OS device. This model serves as a blueprint for the data we expect from the Google Admin SDK responses. Welcome to Pydantic — Pydantic is a data validation and settings management library using Python type annotations. Pydantic ensures that the data you work with adheres to the format you expect and helps in catching errors early in the development process by validating data types and content automatically. Here’s a breakdown of the model I made and its components:

class Status(Enum):
    """Status Enum represents the status of a Chrome OS device."""

    ACTIVE = "ACTIVE"
    DELINQUENT = "DELINQUENT"
    PRE_PROVISIONED = "PRE_PROVISIONED"
    DEPROVISIONED = "DEPROVISIONED"
    DISABLED = "DISABLED"
    INACTIVE = "INACTIVE"
    RETURN_ARRIVED = "RETURN_ARRIVED"
    RETURN_REQUESTED = "RETURN_REQUESTED"
    UNKNOWN = "UNKNOWN"

class RecentUser(PydanticBaseModelConfig):
    type: Literal["USER_TYPE_MANAGED", "USER_TYPE_UNMANAGED"]
    email: str

class ActiveTimeRange(PydanticBaseModelConfig):
    activeTime: int | None = None  # Duration of usage in milliseconds.
    activeTimeSec: int | None = None  # Duration of usage in seconds.
    date: datetime | None = None  # Date of usage.

class ChromeOSDevice(PydanticBaseModelConfig):
    """Chrome OS Device Model represents the details of a Chrome OS device."""

    model_config = ConfigDict(extra="ignore")

    deviceId: str
    serialNumber: Annotated[str | None, StringConstraints(to_upper=True)] = None
    status: Status
    lastSync: Optional[datetime] = None
    supportEndDate: Optional[datetime] = None
    annotatedUser: Optional[str] = None
    annotatedLocation: Optional[str] = None
    notes: Optional[str] = None
    model: Optional[str] = None
    orderNumber: Optional[str] = None
    willAutoRenew: Optional[bool] = None
    osVersion: Optional[str] = None
    platformVersion: Optional[str] = None
    firmwareVersion: Optional[str] = None
    macAddress: Optional[str] = None
    bootMode: Optional[str] = None
    lastEnrollmentTime: Optional[datetime] = None

    # Recent Users
    recentUsers: list[RecentUser] = Field(
        default_factory=list,
        description="List of recent users, ordered by last login time with the most recent first. The number of users listed can vary.",
    )
    lastActiveUserEmail: Optional[str] = None  # EXTRA FIELD

    # Active Time
    activeTimeRanges: list[ActiveTimeRange] = Field(
        default_factory=list,
        description="List of active time ranges. If the device is active, this list will be non-empty.",
    )

    ethernetMacAddress: Optional[str] = None
    annotatedAssetId: Optional[str] = None

    totalStorageSpaceInBytes: Optional[int] = 0
    freeStorageSpaceInBytes: Optional[int] = 0
    totalStorageSpaceInGBs: Optional[float] = 0
    freeStorageSpaceInGBs: Optional[float] = 0

    systemRamTotalBytes: Optional[int] = 0
    systemRamFreeBytes: Optional[int] = 0
    systemRamTotalGBs: Optional[float] = 0
    systemRamFreeGBs: Optional[float] = 0

    localIPAddress: Optional[str] = None
    wanIPAddress: Optional[str] = None

    autoUpdateExpiration: Optional[datetime] = None
    manufactureDate: Optional[date] = None
    orgUnitPath: Optional[str] = None
    orgUnitId: Optional[str] = None

    firstEnrollmentTime: Optional[datetime] = None
    lastDeprovisionTimestamp: Optional[datetime] = None
    deprovisionReason: Optional[str] = None

    deviceLicenseType: Optional[str] = None

    # WAN IP Address Info # EXTRA FIELDS
    wan_ip_lat: Optional[float] = None
    wan_ip_lon: Optional[float] = None
    wan_ip_isp: Optional[str] = None
    wan_ip_city: Optional[str] = None
    wan_ip_regionName: Optional[str] = None

    @model_validator(mode="before")
    @classmethod
    def fine_tune_model(cls, values):
        """Fine-tune the model before validation."""
        # Calculate total and free storage space in GBs
        disk_volume_report = values.get("diskVolumeReports", [])
        if len(disk_volume_report) > 0:
            volume_list = disk_volume_report[0].get("volumeInfo", [])
            for volume in volume_list:
                if "/home" in volume.get("volumeId"):
                    values["totalStorageSpaceInBytes"] = volume.get("storageTotal")
                    values["freeStorageSpaceInBytes"] = volume.get("storageFree")
                    values["totalStorageSpaceInGBs"] = round(
                        int(volume.get("storageTotal")) / 1024**3, 2
                    )
                    values["freeStorageSpaceInGBs"] = round(
                        int(volume.get("storageFree")) / 1024**3, 2
                    )
                    break

        # Update the Last Active User Email
        if values.get("recentUsers"):
            values["lastActiveUserEmail"] = values["recentUsers"][0].get("email")

        # Update the System RAM Total and Free in GBs
        values["systemRamTotalBytes"] = values.get("systemRamTotalBytes", 0)
        values["systemRamTotalGBs"] = round(values["systemRamTotalBytes"] / 1024**3, 2)
        system_ram_free_list = values.get("systemRamFreeReports", [])
        if system_ram_free_list:
            free_ram_bytes = system_ram_free_list[-1].get("systemRamFreeInfo", 0)[0]
            values["systemRamFreeBytes"] = int(free_ram_bytes)
            values["systemRamFreeGBs"] = round(int(free_ram_bytes) / 1024**3, 2)

        # lastKnownNetwork is a list of dictionaries
        last_known_network = values.get("lastKnownNetwork", [])
        if last_known_network:
            values["localIPAddress"] = last_known_network[0].get("ipAddress")
            values["wanIPAddress"] = last_known_network[0].get("wanIpAddress")
            # Get IP address from the last known network
            result_ip = get_ip_info(values["wanIPAddress"])
            if result_ip:
                values["wan_ip_lat"] = result_ip.lat
                values["wan_ip_lon"] = result_ip.lon
                values["wan_ip_isp"] = result_ip.isp
                values["wan_ip_city"] = result_ip.city
                values["wan_ip_regionName"] = result_ip.regionName
        # Convert the autoUpdateExpiration from unix timestamp to datetime
        auto_update_expiration = values.get("autoUpdateExpiration")
        if auto_update_expiration:
            values["autoUpdateExpiration"] = datetime.fromtimestamp(
                auto_update_expiration
            )
        # convert activeTimeRanges milliseconds to seconds
        active_time_ranges = values.get("activeTimeRanges", [])
        for active_time_range in active_time_ranges:
            active_time_range["activeTimeSec"] = int(
                active_time_range.get("activeTime") / 1000
            )
        return values

Conclusion and Next Steps

We’ve covered a lot in this guide — from setting up authentication with Google Workspace to fetching Chrome OS devices using the Google Admin SDK and structuring our data with Pydantic models. I hope this blog has simplified some of the complexities you might face while working with Google APIs and has provided you with actionable steps to integrate these tools into your projects effectively.

Thanks for reading! 🌟

Stay connected and follow my journey:

• *Linktree: Find all my social media links and more.*
• *Website: Dive deeper into my projects and portfolio.*

Feel free to leave your thoughts in the comments or reach out on any of my profiles. I look forward to hearing from you!