AWS Lambda with Python

viio-sync-api-sdk

import asyncio
import json
import os
from typing import Any
from urllib.parse import urlencode, urlsplit
from urllib.request import Request, urlopen

from viio_sync_api import SyncApiAuthentication, SyncApiClient, SyncApiClientOptions
from viio_sync_api.v1 import (
    Employee,
    EmployeeRecords,
    MicrosoftEmployeeDetails,
    MicrosoftEmployeeOrgData,
    SyncRecordsRequest,
)


GRAPH_USERS_URL = "https://graph.microsoft.com/v1.0/users?" + urlencode(
    {
        "$select": ",".join(
            (
                "id",
                "displayName",
                "givenName",
                "surname",
                "mail",
                "userPrincipalName",
                "department",
                "accountEnabled",
                "jobTitle",
                "userType",
                "country",
                "employeeOrgData",
            )
        ),
        "$top": "999",
    }
)


def request_json(request: Request) -> dict[str, Any]:
    with urlopen(request, timeout=30) as response:
        payload = json.load(response)
    if not isinstance(payload, dict):
        raise RuntimeError("Remote service returned an invalid JSON response")
    return payload


def request_graph_token() -> str:
    tenant_id = os.environ["MICROSOFT_TENANT_ID"]
    allowed_characters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-"
    if not tenant_id or any(character not in allowed_characters for character in tenant_id):
        raise ValueError("MICROSOFT_TENANT_ID must be a tenant ID or verified domain name")

    request = Request(
        f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token",
        data=urlencode(
            {
                "grant_type": "client_credentials",
                "client_id": os.environ["MICROSOFT_CLIENT_ID"],
                "client_secret": os.environ["MICROSOFT_CLIENT_SECRET"],
                "scope": "https://graph.microsoft.com/.default",
            }
        ).encode(),
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        method="POST",
    )
    token = request_json(request).get("access_token")
    if not isinstance(token, str) or not token:
        raise RuntimeError("Microsoft identity platform returned no access token")
    return token


def request_graph_users(url: str, access_token: str) -> tuple[list[dict[str, Any]], str | None]:
    parsed_url = urlsplit(url)
    if parsed_url.scheme != "https" or parsed_url.hostname != "graph.microsoft.com":
        raise RuntimeError("Microsoft Graph returned an invalid pagination URL")

    payload = request_json(
        Request(
            url,
            headers={
                "Accept": "application/json",
                "Authorization": f"Bearer {access_token}",
            },
        )
    )
    users = payload.get("value")
    if not isinstance(users, list) or not all(isinstance(user, dict) for user in users):
        raise RuntimeError("Microsoft Graph returned an invalid users response")

    next_url = payload.get("@odata.nextLink")
    if next_url is not None and not isinstance(next_url, str):
        raise RuntimeError("Microsoft Graph returned an invalid pagination URL")
    return users, next_url


def required_string(source: dict[str, Any], field: str) -> str:
    value = source.get(field)
    if not isinstance(value, str) or not value:
        raise RuntimeError(f"Microsoft Graph user is missing {field}")
    return value


def optional_string(source: dict[str, Any], field: str) -> str | None:
    value = source.get(field)
    return value if isinstance(value, str) and value else None


def optional_bool(source: dict[str, Any], field: str) -> bool | None:
    value = source.get(field)
    return value if isinstance(value, bool) else None


def employee_org_data(user: dict[str, Any]) -> MicrosoftEmployeeOrgData | None:
    value = user.get("employeeOrgData")
    if not isinstance(value, dict):
        return None

    division = optional_string(value, "division")
    cost_center = optional_string(value, "costCenter")
    if division is None and cost_center is None:
        return None

    return MicrosoftEmployeeOrgData(division=division, cost_center=cost_center)


def employee_from_graph_user(user: dict[str, Any]) -> Employee:
    return Employee(
        microsoft=MicrosoftEmployeeDetails(
            id=required_string(user, "id"),
            display_name=required_string(user, "displayName"),
            given_name=optional_string(user, "givenName"),
            surname=optional_string(user, "surname"),
            mail=optional_string(user, "mail"),
            user_principal_name=required_string(user, "userPrincipalName"),
            department=optional_string(user, "department"),
            account_enabled=optional_bool(user, "accountEnabled"),
            job_title=optional_string(user, "jobTitle"),
            user_type=optional_string(user, "userType"),
            country=optional_string(user, "country"),
            employee_org_data=employee_org_data(user),
        )
    )


async def sync_employees() -> int:
    graph_token = await asyncio.to_thread(request_graph_token)
    options = SyncApiClientOptions(
        endpoint=os.environ["VIIO_SYNC_API_ENDPOINT"],
        authentication=SyncApiAuthentication.api_key(os.environ["VIIO_SYNC_API_KEY"]),
    )

    async with SyncApiClient(options) as client:
        batch = await client.start_batch(os.environ["VIIO_DIRECT_INTEGRATION_INSTALLATION_ID"])
        employee_count = 0
        next_url: str | None = GRAPH_USERS_URL
        try:
            while next_url is not None:
                users, next_url = await asyncio.to_thread(request_graph_users, next_url, graph_token)
                if users:
                    employees = [employee_from_graph_user(user) for user in users]
                    await batch.write(SyncRecordsRequest(employees=EmployeeRecords(records=employees)))
                    employee_count += len(employees)

            if employee_count == 0:
                await batch.write(SyncRecordsRequest(employees=EmployeeRecords()))
            await batch.complete()
        except BaseException:
            await batch.abort()
            raise
        return employee_count


def lambda_handler(event: dict[str, Any], context: object) -> dict[str, Any]:
    del event, context
    employee_count = asyncio.run(sync_employees())
    return {"ok": True, "employee_count": employee_count}

Resources:
  EmployeeSyncFunction:
    Type: AWS::Serverless::Function
    Properties:
      Architectures:
        - arm64
      CodeUri: .
      Handler: aws_lambda.lambda_handler
      Runtime: python3.11
      MemorySize: 512
      Timeout: 900
      Environment:
        Variables:
          MICROSOFT_TENANT_ID: !Ref MicrosoftTenantId
          MICROSOFT_CLIENT_ID: !Ref MicrosoftClientId
          MICROSOFT_CLIENT_SECRET: !Ref MicrosoftClientSecret
          VIIO_SYNC_API_ENDPOINT: !Ref ViioSyncApiEndpoint
          VIIO_SYNC_API_KEY: !Ref ViioSyncApiKey
          VIIO_DIRECT_INTEGRATION_INSTALLATION_ID: !Ref ViioDirectIntegrationInstallationId
      Events:
        NightlyEmployeeSync:
          Type: ScheduleV2
          Properties:
            ScheduleExpression: cron(0 2 * * ? *)
            ScheduleExpressionTimezone: UTC

python -m pip install \
  --platform manylinux2014_aarch64 \
  --implementation cp \
  --python-version 3.11 \
  --only-binary=:all: \
  --target package \
  viio-sync-api-sdk