graph

napt.graph.client

Microsoft Graph HTTP transport shared by every Graph caller in NAPT.

Provides graph_request, the single function through which Intune app management, assignment, and app registration calls reach Graph, along with the header builders callers pass to it. Endpoint-specific wrappers live in napt.graph.intune and napt.auth.registration.

Graph calls retry transient failures -- HTTP 429 (honoring Retry-After), transient server errors, and connection drops -- with bounded exponential backoff before raising. Resource-creating POSTs retry only unambiguous throttling responses, so a lost reply to a processed create is never resubmitted as a duplicate.

auth_headers

auth_headers(access_token: str) -> dict[str, str]

Returns the Authorization header for a bodiless Graph request.

Parameters:

Name	Type	Description	Default
access_token	str	Bearer token for Graph API.	required

Returns:

Type	Description
dict[str, str]	Headers carrying the bearer token.

Source code in napt/graph/client.py

def auth_headers(access_token: str) -> dict[str, str]:
    """Returns the Authorization header for a bodiless Graph request.

    Args:
        access_token: Bearer token for Graph API.

    Returns:
        Headers carrying the bearer token.

    """
    return {"Authorization": f"Bearer {access_token}"}

json_headers

json_headers(access_token: str) -> dict[str, str]

Returns the headers for a Graph request with a JSON body.

Parameters:

Name	Type	Description	Default
access_token	str	Bearer token for Graph API.	required

Returns:

Type	Description
dict[str, str]	Headers carrying the bearer token and a JSON content type.

Source code in napt/graph/client.py

def json_headers(access_token: str) -> dict[str, str]:
    """Returns the headers for a Graph request with a JSON body.

    Args:
        access_token: Bearer token for Graph API.

    Returns:
        Headers carrying the bearer token and a JSON content type.

    """
    return {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
    }

graph_request

graph_request(
    method: str,
    url: str,
    context: str,
    headers: dict[str, str],
    json: dict | None = None,
    ok_statuses: tuple[int, ...] = (),
    idempotent: bool = True,
    deadline: float | None = None,
) -> dict

Issues a Graph API request, retrying transient failures.

HTTP 429 (honoring Retry-After) and transient server errors retry with exponential backoff, as do connection-level failures. Non-idempotent calls (resource-creating POSTs) retry only statuses that guarantee the request was shed before processing, and never connection failures — a lost reply to a processed create must not be resubmitted. Every other response is checked immediately, so permission and validation errors surface without retrying. The last attempt's failure is raised with full response detail. Each request carries a fresh client-request-id header for Microsoft support correlation.

Parameters:

Name	Type	Description	Default
method	str	HTTP method name.	required
url	str	Full request URL.	required
context	str	Short description of the operation for error messages.	required
headers	dict[str, str]	Request headers, including authorization.	required
json	dict | None	Optional JSON body.	None
ok_statuses	tuple[int, ...]	Statuses to treat as success with an empty body (e.g. 404 for an idempotent delete).	()
idempotent	bool	Whether resubmitting this request is always safe. False restricts retries to unambiguous throttling responses.	True
deadline	float | None	Optional time.monotonic() budget; a retry wait that would run past it surfaces the failure instead.	None

Returns:

Type	Description
dict	Parsed JSON body as a dict, or empty dict for empty responses
dict	and ok_statuses matches.

Raises:

Type	Description
AuthError	On 401 or 403.
ConfigError	On 400.
NetworkError	On any other non-2xx status once retries are exhausted, or on a connection failure.

Source code in napt/graph/client.py

def graph_request(
    method: str,
    url: str,
    context: str,
    headers: dict[str, str],
    json: dict | None = None,
    ok_statuses: tuple[int, ...] = (),
    idempotent: bool = True,
    deadline: float | None = None,
) -> dict:
    """Issues a Graph API request, retrying transient failures.

    HTTP 429 (honoring ``Retry-After``) and transient server errors
    retry with exponential backoff, as do connection-level failures.
    Non-idempotent calls (resource-creating POSTs) retry only statuses
    that guarantee the request was shed before processing, and never
    connection failures — a lost reply to a processed create must not
    be resubmitted. Every other response is checked immediately, so
    permission and validation errors surface without retrying. The last
    attempt's failure is raised with full response detail. Each request
    carries a fresh ``client-request-id`` header for Microsoft support
    correlation.

    Args:
        method: HTTP method name.
        url: Full request URL.
        context: Short description of the operation for error messages.
        headers: Request headers, including authorization.
        json: Optional JSON body.
        ok_statuses: Statuses to treat as success with an empty body
            (e.g. 404 for an idempotent delete).
        idempotent: Whether resubmitting this request is always safe.
            False restricts retries to unambiguous throttling responses.
        deadline: Optional ``time.monotonic()`` budget; a retry wait
            that would run past it surfaces the failure instead.

    Returns:
        Parsed JSON body as a dict, or empty dict for empty responses
        and ``ok_statuses`` matches.

    Raises:
        AuthError: On 401 or 403.
        ConfigError: On 400.
        NetworkError: On any other non-2xx status once retries are
            exhausted, or on a connection failure.

    """
    from napt.logging import get_global_logger

    logger = get_global_logger()
    retry_statuses = (
        _GRAPH_RETRY_STATUS if idempotent else _GRAPH_RETRY_STATUS_UNAMBIGUOUS
    )
    delay = _GRAPH_RETRY_INITIAL_DELAY
    for attempt in range(1, _GRAPH_RETRY_ATTEMPTS + 1):
        err: Exception | None = None
        resp: requests.Response | None = None
        request_headers = {**headers, "client-request-id": str(uuid.uuid4())}
        try:
            resp = requests.request(
                method, url, headers=request_headers, json=json, timeout=30
            )
        except requests.RequestException as exc:
            if not idempotent:
                # The request may have been processed before the
                # connection died; resubmitting could duplicate the
                # resource. Surface it — re-running converges through
                # the flow-level stamp adoption.
                raise NetworkError(f"{context}: {exc}") from exc
            err = exc
            detail = str(exc)
        else:
            if resp.status_code in ok_statuses:
                return {}
            if resp.status_code not in retry_statuses:
                return _check_response(resp, context)
            detail = f"HTTP {resp.status_code}"

        if attempt == _GRAPH_RETRY_ATTEMPTS:
            if resp is not None:
                return _check_response(resp, context)
            raise NetworkError(
                f"{context} after {_GRAPH_RETRY_ATTEMPTS} attempts: {detail}"
            ) from err

        wait = _retry_wait(resp, delay)
        if deadline is not None and time.monotonic() + wait >= deadline:
            # No budget left for another attempt; surface this failure.
            if resp is not None:
                return _check_response(resp, context)
            raise NetworkError(f"{context}: {detail}") from err
        logger.warning(
            "HTTP",
            f"{context}: transient failure ({detail}); retrying in "
            f"{wait:.0f}s (attempt {attempt}/{_GRAPH_RETRY_ATTEMPTS})",
        )
        time.sleep(wait)
        delay *= 2

    raise NetworkError(f"{context}: retry attempts exhausted")  # pragma: no cover

napt.graph.intune

Intune app management calls: Win32 app upload, queries, and assignments.

Implements the full upload flow for a Win32 LOB app:

1. Create Win32 app record in Intune (POST mobileApps)
2. Create a content version (POST contentVersions)
3. Create a file entry and wait for SAS URI (POST files + polling)
4. Upload encrypted payload to Azure Blob Storage (PUT blocks + block list)
5. Commit the uploaded file with encryption metadata (POST commit + polling)
6. Set the committed content version on the app (PATCH mobileApps)

Also provides app queries used for reconciliation (list_mobile_apps, get_mobile_app, update_win32_app) and group-based assignment plumbing (resolve_group_id, resolve_assignment_target, get_app_assignments, build_assignment, assign_app) used by deployment promotion.

All functions take an access_token as the first argument. Obtain one via get_access_token. Graph calls go through graph_request and inherit its retry behavior; Azure Blob PUTs carry their own retry tuned for SAS-propagation 403s.

resolve_group_id

resolve_group_id(access_token: str, group: str) -> str

Resolves an Entra ID group name or object ID to an object ID.

Values that already look like object IDs (GUIDs) pass through without a Graph call. Names are looked up by exact displayName match, which requires the Group.Read.All application permission.

Parameters:

Name	Type	Description	Default
access_token	str	Bearer token for Graph API.	required
group	str	Group displayName or object ID (GUID).	required

Returns:

Type	Description
str	The group's object ID.

Raises:

Type	Description
AuthError	On 401 or 403 (check Group.Read.All permission).
ConfigError	If no group or more than one group matches the name.
NetworkError	On 5xx or connection error.

Source code in napt/graph/intune.py

def resolve_group_id(access_token: str, group: str) -> str:
    """Resolves an Entra ID group name or object ID to an object ID.

    Values that already look like object IDs (GUIDs) pass through without
    a Graph call. Names are looked up by exact displayName match, which
    requires the Group.Read.All application permission.

    Args:
        access_token: Bearer token for Graph API.
        group: Group displayName or object ID (GUID).

    Returns:
        The group's object ID.

    Raises:
        AuthError: On 401 or 403 (check Group.Read.All permission).
        ConfigError: If no group or more than one group matches the name.
        NetworkError: On 5xx or connection error.

    """
    if _GUID_RE.match(group):
        return group

    # OData doubles the quote; the URL then needs the rest percent-encoded,
    # or a name with "&" or "#" ends the filter early.
    escaped = quote(group.replace("'", "''"), safe="'")
    url = (
        f"{GRAPH_BASE}/groups"
        f"?$filter=displayName eq '{escaped}'&$select=id,displayName"
    )
    body = graph_request(
        "GET", url, "resolve_group_id", headers=auth_headers(access_token)
    )
    matches: list[dict] = body.get("value", [])

    if not matches:
        raise ConfigError(
            f"No Entra ID group found with displayName '{group}'. "
            "Check the name, or use the group's object ID instead."
        )
    if len(matches) > 1:
        ids = ", ".join(m["id"] for m in matches)
        raise ConfigError(
            f"Multiple Entra ID groups share the displayName '{group}' "
            f"({ids}). Use the object ID of the intended group instead."
        )
    return matches[0]["id"]

resolve_assignment_target

resolve_assignment_target(
    access_token: str, group: str, group_id_cache: dict[str, str] | None = None
) -> dict

Resolves a deployment group entry to an assignment target dict.

The reserved names "All Users" and "All Devices" map to Intune's built-in virtual targets; anything else resolves to an Entra ID group target via resolve_group_id.

Parameters:

Name	Type	Description	Default
access_token	str	Bearer token for Graph API.	required
group	str	Group displayName, object ID, or reserved virtual name.	required
group_id_cache	dict[str, str] | None	Optional cache of name to object ID, shared across calls to avoid repeated lookups.	None

Returns:

Type	Description
dict	An assignment target dict for use with build_assignment.

Raises:

Type	Description
AuthError	On 401 or 403 (check Group.Read.All permission).
ConfigError	If no group or more than one group matches a name.
NetworkError	On 5xx or connection error.

Source code in napt/graph/intune.py

def resolve_assignment_target(
    access_token: str,
    group: str,
    group_id_cache: dict[str, str] | None = None,
) -> dict:
    """Resolves a deployment group entry to an assignment target dict.

    The reserved names "All Users" and "All Devices" map to Intune's
    built-in virtual targets; anything else resolves to an Entra ID
    group target via resolve_group_id.

    Args:
        access_token: Bearer token for Graph API.
        group: Group displayName, object ID, or reserved virtual name.
        group_id_cache: Optional cache of name to object ID, shared
            across calls to avoid repeated lookups.

    Returns:
        An assignment target dict for use with build_assignment.

    Raises:
        AuthError: On 401 or 403 (check Group.Read.All permission).
        ConfigError: If no group or more than one group matches a name.
        NetworkError: On 5xx or connection error.

    """
    if group in VIRTUAL_TARGETS:
        return dict(VIRTUAL_TARGETS[group])
    if group_id_cache is None:
        group_id_cache = {}
    if group not in group_id_cache:
        group_id_cache[group] = resolve_group_id(access_token, group)
    return {
        "@odata.type": "#microsoft.graph.groupAssignmentTarget",
        "groupId": group_id_cache[group],
    }

get_app_assignments

get_app_assignments(access_token: str, app_id: str) -> list[dict]

Gets the current assignments of a mobile app.

Parameters:

Name	Type	Description	Default
access_token	str	Bearer token for Graph API.	required
app_id	str	Graph API object ID of the app.	required

Returns:

Type	Description
list[dict]	A list of mobileAppAssignment dicts (empty when unassigned).

Raises:

Type	Description
AuthError	On 401 or 403.
NetworkError	On 5xx or connection error.

Source code in napt/graph/intune.py

def get_app_assignments(access_token: str, app_id: str) -> list[dict]:
    """Gets the current assignments of a mobile app.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the app.

    Returns:
        A list of mobileAppAssignment dicts (empty when unassigned).

    Raises:
        AuthError: On 401 or 403.
        NetworkError: On 5xx or connection error.

    """
    url = f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}/assignments"
    body = graph_request(
        "GET", url, "get_app_assignments", headers=auth_headers(access_token)
    )
    return body.get("value", [])

build_assignment

build_assignment(target: dict, intent: str) -> dict

Builds a mobileAppAssignment payload for a resolved target.

Parameters:

Name	Type	Description	Default
target	dict	An assignment target dict (group or virtual target).	required
intent	str	Assignment intent, "available" or "required".	required

Returns:

Type	Description
dict	A mobileAppAssignment dict for use with assign_app.

Source code in napt/graph/intune.py

def build_assignment(target: dict, intent: str) -> dict:
    """Builds a mobileAppAssignment payload for a resolved target.

    Args:
        target: An assignment target dict (group or virtual target).
        intent: Assignment intent, "available" or "required".

    Returns:
        A mobileAppAssignment dict for use with assign_app.

    """
    return {
        "@odata.type": "#microsoft.graph.mobileAppAssignment",
        "intent": intent,
        "target": target,
    }

assign_app

assign_app(access_token: str, app_id: str, assignments: list[dict]) -> None

Sets a mobile app's assignments.

The assign action replaces the app's entire assignment set. Callers that intend to preserve existing assignments must read them first with get_app_assignments and include them in the new list.

Parameters:

Name	Type	Description	Default
access_token	str	Bearer token for Graph API.	required
app_id	str	Graph API object ID of the app.	required
assignments	list[dict]	Complete list of mobileAppAssignment dicts to apply.	required

Raises:

Type	Description
AuthError	On 401 or 403.
ConfigError	On 400 (invalid assignment payload).
NetworkError	On 5xx or connection error.

Source code in napt/graph/intune.py

def assign_app(access_token: str, app_id: str, assignments: list[dict]) -> None:
    """Sets a mobile app's assignments.

    The assign action replaces the app's entire assignment set. Callers
    that intend to preserve existing assignments must read them first with
    get_app_assignments and include them in the new list.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the app.
        assignments: Complete list of mobileAppAssignment dicts to apply.

    Raises:
        AuthError: On 401 or 403.
        ConfigError: On 400 (invalid assignment payload).
        NetworkError: On 5xx or connection error.

    """
    url = f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}/assign"
    body = {"mobileAppAssignments": assignments}
    graph_request(
        "POST", url, "assign_app", headers=json_headers(access_token), json=body
    )

list_mobile_apps

list_mobile_apps(access_token: str) -> list[dict]

Lists all mobile apps in the tenant with id, displayName, and notes.

Follows @odata.nextLink pagination until the collection is exhausted.

Parameters:

Name	Type	Description	Default
access_token	str	Bearer token for Graph API.	required

Returns:

Type	Description
list[dict]	A list of app dicts, each with at least "id", "displayName", and "notes" keys.

Raises:

Type	Description
AuthError	On 401 or 403.
NetworkError	On 5xx or connection error.

Source code in napt/graph/intune.py

def list_mobile_apps(access_token: str) -> list[dict]:
    """Lists all mobile apps in the tenant with id, displayName, and notes.

    Follows @odata.nextLink pagination until the collection is exhausted.

    Args:
        access_token: Bearer token for Graph API.

    Returns:
        A list of app dicts, each with at least "id", "displayName", and
            "notes" keys.

    Raises:
        AuthError: On 401 or 403.
        NetworkError: On 5xx or connection error.

    """
    url: str | None = (
        f"{GRAPH_BASE}/deviceAppManagement/mobileApps?$select=id,displayName,notes"
    )
    apps: list[dict] = []
    while url:
        body = graph_request(
            "GET", url, "list_mobile_apps", headers=auth_headers(access_token)
        )
        apps.extend(body.get("value", []))
        url = body.get("@odata.nextLink")
    return apps

delete_mobile_app

delete_mobile_app(access_token: str, app_id: str) -> None

Deletes a mobile app from Intune.

A 404 is tolerated — the app being already gone is the desired end state, so retried deletions stay idempotent.

Parameters:

Name	Type	Description	Default
access_token	str	Bearer token for Graph API.	required
app_id	str	Graph API object ID of the app to delete.	required

Raises:

Type	Description
AuthError	On 401 or 403.
NetworkError	On 5xx or connection error.

Source code in napt/graph/intune.py

def delete_mobile_app(access_token: str, app_id: str) -> None:
    """Deletes a mobile app from Intune.

    A 404 is tolerated — the app being already gone is the desired end
    state, so retried deletions stay idempotent.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the app to delete.

    Raises:
        AuthError: On 401 or 403.
        NetworkError: On 5xx or connection error.

    """
    url = f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}"
    graph_request(
        "DELETE",
        url,
        "delete_mobile_app",
        headers=auth_headers(access_token),
        ok_statuses=(404,),
    )

get_mobile_app

get_mobile_app(access_token: str, app_id: str) -> dict

Gets one mobile app's full object by Graph API ID.

Used to read subtype fields that $select on the collection cannot reliably return, such as win32LobApp.committedContentVersion.

Parameters:

Name	Type	Description	Default
access_token	str	Bearer token for Graph API.	required
app_id	str	Graph API object ID of the app.	required

Returns:

Type	Description
dict	The full app object dict.

Raises:

Type	Description
AuthError	On 401 or 403.
NetworkError	On 5xx or connection error.

Source code in napt/graph/intune.py

def get_mobile_app(access_token: str, app_id: str) -> dict:
    """Gets one mobile app's full object by Graph API ID.

    Used to read subtype fields that $select on the collection cannot
    reliably return, such as win32LobApp.committedContentVersion.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the app.

    Returns:
        The full app object dict.

    Raises:
        AuthError: On 401 or 403.
        NetworkError: On 5xx or connection error.

    """
    url = f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}"
    return graph_request(
        "GET", url, "get_mobile_app", headers=auth_headers(access_token)
    )

create_win32_app

create_win32_app(access_token: str, app_metadata: dict) -> str

Creates a new Win32 LOB app record in Intune.

Parameters:

Name	Type	Description	Default
access_token	str	Bearer token for Graph API.	required
app_metadata	dict	Win32LobApp JSON payload (display name, install commands, detection rules, etc.).	required

Returns:

Type	Description
str	The Graph API object ID of the newly created app.

Raises:

Type	Description
AuthError	On 401 or 403.
ConfigError	On 400 (invalid metadata).
NetworkError	On 5xx or connection error.

Source code in napt/graph/intune.py

def create_win32_app(access_token: str, app_metadata: dict) -> str:
    """Creates a new Win32 LOB app record in Intune.

    Args:
        access_token: Bearer token for Graph API.
        app_metadata: Win32LobApp JSON payload (display name, install
            commands, detection rules, etc.).

    Returns:
        The Graph API object ID of the newly created app.

    Raises:
        AuthError: On 401 or 403.
        ConfigError: On 400 (invalid metadata).
        NetworkError: On 5xx or connection error.

    """
    url = f"{GRAPH_BASE}/deviceAppManagement/mobileApps"
    body = graph_request(
        "POST",
        url,
        "create_win32_app",
        headers=json_headers(access_token),
        json=app_metadata,
        idempotent=False,
    )
    return body["id"]

update_win32_app

update_win32_app(access_token: str, app_id: str, app_metadata: dict) -> None

Updates an existing Win32 LOB app record's metadata in Intune.

Parameters:

Name	Type	Description	Default
access_token	str	Bearer token for Graph API.	required
app_id	str	Graph API object ID of the app to update.	required
app_metadata	dict	Win32LobApp JSON payload to apply.	required

Raises:

Type	Description
AuthError	On 401 or 403.
ConfigError	On 400 (invalid metadata).
NetworkError	On 5xx or connection error.

Source code in napt/graph/intune.py

def update_win32_app(access_token: str, app_id: str, app_metadata: dict) -> None:
    """Updates an existing Win32 LOB app record's metadata in Intune.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the app to update.
        app_metadata: Win32LobApp JSON payload to apply.

    Raises:
        AuthError: On 401 or 403.
        ConfigError: On 400 (invalid metadata).
        NetworkError: On 5xx or connection error.

    """
    url = f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}"
    graph_request(
        "PATCH",
        url,
        "update_win32_app",
        headers=json_headers(access_token),
        json=app_metadata,
    )

create_content_version

create_content_version(access_token: str, app_id: str) -> str

Creates a new content version for a Win32 app.

Parameters:

Name	Type	Description	Default
access_token	str	Bearer token for Graph API.	required
app_id	str	Graph API object ID of the Win32 app.	required

Returns:

Type	Description
str	The content version ID string.

Raises:

Type	Description
AuthError	On 401 or 403.
NetworkError	On 5xx or connection error.

Source code in napt/graph/intune.py

def create_content_version(access_token: str, app_id: str) -> str:
    """Creates a new content version for a Win32 app.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the Win32 app.

    Returns:
        The content version ID string.

    Raises:
        AuthError: On 401 or 403.
        NetworkError: On 5xx or connection error.

    """
    url = (
        f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}"
        f"/microsoft.graph.win32LobApp/contentVersions"
    )
    body = graph_request(
        "POST",
        url,
        "create_content_version",
        headers=json_headers(access_token),
        json={},
        idempotent=False,
    )
    return body["id"]

create_content_version_file

create_content_version_file(
    access_token: str, app_id: str, cv_id: str, metadata: IntunewinMetadata
) -> tuple[str, str]

Creates a file entry for a content version and waits for the SAS URI.

Posts the file size information to Graph API, then polls until Azure Storage has provisioned a SAS URI for the upload.

Parameters:

Name	Type	Description	Default
access_token	str	Bearer token for Graph API.	required
app_id	str	Graph API object ID of the Win32 app.	required
cv_id	str	Content version ID from create_content_version.	required
metadata	IntunewinMetadata	Parsed .intunewin metadata (provides file sizes).	required

Returns:

Type	Description
tuple[str, str]	A tuple of (file_id, sas_uri) where sas_uri is the Azure Blob Storage SAS URI to upload the encrypted payload to.

Raises:

Type	Description
AuthError	On 401 or 403.
NetworkError	On 5xx, connection error, or upload state error.

Source code in napt/graph/intune.py

def create_content_version_file(
    access_token: str,
    app_id: str,
    cv_id: str,
    metadata: IntunewinMetadata,
) -> tuple[str, str]:
    """Creates a file entry for a content version and waits for the SAS URI.

    Posts the file size information to Graph API, then polls until Azure
    Storage has provisioned a SAS URI for the upload.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the Win32 app.
        cv_id: Content version ID from create_content_version.
        metadata: Parsed .intunewin metadata (provides file sizes).

    Returns:
        A tuple of (file_id, sas_uri) where sas_uri is the Azure Blob
            Storage SAS URI to upload the encrypted payload to.

    Raises:
        AuthError: On 401 or 403.
        NetworkError: On 5xx, connection error, or upload state error.

    """
    base_url = (
        f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}"
        f"/microsoft.graph.win32LobApp/contentVersions/{cv_id}/files"
    )
    body = {
        "@odata.type": "#microsoft.graph.mobileAppContentFile",
        "name": metadata.encrypted_file_name,
        "size": metadata.unencrypted_content_size,
        "sizeEncrypted": metadata.encrypted_file_size,
        "manifest": None,
        "isDependency": False,
    }
    file_body = graph_request(
        "POST",
        base_url,
        "create_content_version_file",
        headers=json_headers(access_token),
        json=body,
        idempotent=False,
    )
    file_id: str = file_body["id"]

    poll_url = f"{base_url}/{file_id}"
    data = _poll(
        access_token,
        poll_url,
        success_state="azureStorageUriRequestSuccess",
        context="create_content_version_file (poll SAS URI)",
    )
    return file_id, data["azureStorageUri"]

upload_to_azure_blob

upload_to_azure_blob(sas_uri: str, encrypted_payload_path: Path) -> None

Uploads the encrypted payload to Azure Blob Storage using block blobs.

Splits the file into CHUNK_SIZE chunks, uploads each as a block with a base64-encoded block ID, then commits the block list. Prints an inline progress percentage as each chunk completes. Transient per-request failures (including 403 from a not-yet-propagated SAS URI) are retried with backoff.

Parameters:

Name	Type	Description	Default
sas_uri	str	Azure Blob Storage SAS URI from create_content_version_file.	required
encrypted_payload_path	Path	Path to the extracted encrypted payload file (IntunePackage.intunewin from inside the .intunewin ZIP).	required

Raises:

Type	Description
NetworkError	If any block upload or the block list commit fails after retries.

Source code in napt/graph/intune.py

def upload_to_azure_blob(
    sas_uri: str,
    encrypted_payload_path: Path,
) -> None:
    """Uploads the encrypted payload to Azure Blob Storage using block blobs.

    Splits the file into CHUNK_SIZE chunks, uploads each as a block with a
    base64-encoded block ID, then commits the block list. Prints an inline
    progress percentage as each chunk completes. Transient per-request
    failures (including 403 from a not-yet-propagated SAS URI) are retried
    with backoff.

    Args:
        sas_uri: Azure Blob Storage SAS URI from create_content_version_file.
        encrypted_payload_path: Path to the extracted encrypted payload file
            (IntunePackage.intunewin from inside the .intunewin ZIP).

    Raises:
        NetworkError: If any block upload or the block list commit fails
            after retries.

    """
    from napt.logging import get_global_logger

    logger = get_global_logger()

    block_ids: list[str] = []
    total_bytes = encrypted_payload_path.stat().st_size
    bytes_uploaded = 0
    last_percent = -1

    started_at = time.time()
    with open(encrypted_payload_path, "rb") as fh:
        block_index = 0
        while True:
            chunk = fh.read(CHUNK_SIZE)
            if not chunk:
                break

            # Block ID: base64(zero-padded 5-digit decimal index)
            block_id = base64.b64encode(str(block_index).zfill(5).encode()).decode()
            block_ids.append(block_id)

            put_url = f"{sas_uri}&comp=block&blockid={block_id}"
            _blob_put_with_retry(
                put_url,
                chunk,
                headers={
                    "x-ms-blob-type": "BlockBlob",
                    "Content-Length": str(len(chunk)),
                },
                context=f"Azure Blob block upload failed (block {block_index})",
            )

            bytes_uploaded += len(chunk)
            if total_bytes:
                pct = int(bytes_uploaded * 100 / total_bytes)
                if pct != last_percent:
                    logger.progress("UPLOAD", f"{pct}%")
                    last_percent = pct

            block_index += 1

    # Commit all blocks by submitting the block list
    block_list_xml = (
        '<?xml version="1.0" encoding="utf-8"?>\n'
        "<BlockList>\n"
        + "".join(f"  <Latest>{bid}</Latest>\n" for bid in block_ids)
        + "</BlockList>"
    )
    commit_url = f"{sas_uri}&comp=blocklist"
    _blob_put_with_retry(
        commit_url,
        block_list_xml.encode("utf-8"),
        headers={"Content-Type": "application/xml"},
        context="Azure Blob block list commit failed",
        timeout=60,
    )

    elapsed = time.time() - started_at
    speed_mb = (bytes_uploaded / (1024 * 1024)) / elapsed if elapsed > 0 else 0
    size_mb = bytes_uploaded / (1024 * 1024)
    logger.info(
        "UPLOAD",
        f"Complete: {encrypted_payload_path.name} ({size_mb:.1f} MB) "
        f"in {elapsed:.1f}s at {speed_mb:.1f} MB/s",
    )

commit_content_version_file

commit_content_version_file(
    access_token: str,
    app_id: str,
    cv_id: str,
    file_id: str,
    metadata: IntunewinMetadata,
) -> None

Commits the uploaded file with encryption metadata, then waits for confirmation.

Sends the encryption key, MAC, IV, and digest to Graph API, then polls until Intune confirms the file is committed.

Parameters:

Name	Type	Description	Default
access_token	str	Bearer token for Graph API.	required
app_id	str	Graph API object ID of the Win32 app.	required
cv_id	str	Content version ID.	required
file_id	str	File entry ID from create_content_version_file.	required
metadata	IntunewinMetadata	Parsed .intunewin metadata (provides all encryption fields).	required

Raises:

Type	Description
AuthError	On 401 or 403.
NetworkError	On 5xx, connection error, or if commit times out.

Note

Graph returns 200 (not 201) for the commit POST.

Source code in napt/graph/intune.py

def commit_content_version_file(
    access_token: str,
    app_id: str,
    cv_id: str,
    file_id: str,
    metadata: IntunewinMetadata,
) -> None:
    """Commits the uploaded file with encryption metadata, then waits for confirmation.

    Sends the encryption key, MAC, IV, and digest to Graph API, then polls
    until Intune confirms the file is committed.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the Win32 app.
        cv_id: Content version ID.
        file_id: File entry ID from create_content_version_file.
        metadata: Parsed .intunewin metadata (provides all encryption fields).

    Raises:
        AuthError: On 401 or 403.
        NetworkError: On 5xx, connection error, or if commit times out.

    Note:
        Graph returns 200 (not 201) for the commit POST.

    """
    commit_url = (
        f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}"
        f"/microsoft.graph.win32LobApp/contentVersions/{cv_id}"
        f"/files/{file_id}/commit"
    )
    body = {
        "fileEncryptionInfo": {
            "encryptionKey": metadata.encryption_key,
            "macKey": metadata.mac_key,
            "initializationVector": metadata.init_vector,
            "mac": metadata.mac,
            "profileIdentifier": metadata.profile_identifier,
            "fileDigest": metadata.file_digest,
            "fileDigestAlgorithm": metadata.file_digest_algorithm,
        }
    }
    graph_request(
        "POST",
        commit_url,
        "commit_content_version_file",
        headers=json_headers(access_token),
        json=body,
    )

    poll_url = (
        f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}"
        f"/microsoft.graph.win32LobApp/contentVersions/{cv_id}/files/{file_id}"
    )
    _poll(
        access_token,
        poll_url,
        success_state="commitFileSuccess",
        context="commit_content_version_file (poll commit)",
    )

commit_content_version

commit_content_version(access_token: str, app_id: str, cv_id: str) -> None

Sets the committed content version on the Win32 app.

This is the final step — after calling this, the app is fully published in Intune and available for assignment.

Parameters:

Name	Type	Description	Default
access_token	str	Bearer token for Graph API.	required
app_id	str	Graph API object ID of the Win32 app.	required
cv_id	str	Content version ID to mark as committed.	required

Raises:

Type	Description
AuthError	On 401 or 403.
NetworkError	On 5xx or connection error.

Source code in napt/graph/intune.py

def commit_content_version(access_token: str, app_id: str, cv_id: str) -> None:
    """Sets the committed content version on the Win32 app.

    This is the final step — after calling this, the app is fully published
    in Intune and available for assignment.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the Win32 app.
        cv_id: Content version ID to mark as committed.

    Raises:
        AuthError: On 401 or 403.
        NetworkError: On 5xx or connection error.

    """
    url = f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}"
    body = {
        "@odata.type": WIN32_LOB_APP_TYPE,
        "committedContentVersion": cv_id,
    }
    graph_request(
        "PATCH",
        url,
        "commit_content_version",
        headers=json_headers(access_token),
        json=body,
    )