azure-cu-sdk Documentation

Project initialized and authored by: Kintu Sangwan (kintu.sangwn.ref@gmail.com)

This SDK solves the need to reliably submit documents to Azure Content Understanding, poll for completion, and retrieve the exact raw response payload for downstream use.

It provides an async REST client, a helper layer for document analysis, and Pydantic models for typed request and response payloads.

Overview

• azure_cu_sdk.aio: Async import surface for the client and helper utilities.
• azure_cu_sdk.models: Pydantic models for request/response payloads.
• azure_cu_sdk.helpers: Operational helper for document analysis and raw payload retrieval.
• azure_cu_sdk.client_async: Raw async REST client for the Content Understanding API.

Installation

pip install azure-cu-sdk

Quickstart (Async)

import asyncio
from azure_cu_sdk.aio import (
    AsyncAzureContentUnderstandingClient,
    ContentUnderstandingExecutionConfig,
    ContentUnderstandingOperationsHelper,
    DocumentExecutionRequest,
)

async def main() -> None:
    async with AsyncAzureContentUnderstandingClient(
        endpoint="https://example.cognitiveservices.azure.com",
        api_key="...",
        api_version="2025-05-01-preview",
    ) as client:
        helper = ContentUnderstandingOperationsHelper(
            client=client,
            config=ContentUnderstandingExecutionConfig(analyzer_id="my-analyzer"),
        )

        payload = await helper.analyze_document(
            DocumentExecutionRequest(
                document_url="https://example.com/doc.pdf",
                file_name="doc.pdf",
                file_id="doc-001",
            ),
            analyzer_id="override-analyzer",
        )

        print(payload)

asyncio.run(main())

Without a Context Manager

If you do not want to use async with, you must call await client.close() when done:

import asyncio
from azure_cu_sdk.aio import AsyncAzureContentUnderstandingClient

async def main() -> None:
    client = AsyncAzureContentUnderstandingClient(
        endpoint="https://example.cognitiveservices.azure.com",
        api_key="...",
        api_version="2025-05-01-preview",
    )
    try:
        analyzers = await client.get_all_analyzers()
        print(analyzers)
    finally:
        await client.close()

asyncio.run(main())

Package Layout

azure_cu_sdk/
  aio/
    __init__.py
  client.py
  client_async.py
  exceptions.py
  helpers.py
  models.py
  types.py

Async Client (azure_cu_sdk.client_async)

AsyncAzureContentUnderstandingClient

Construct the client with your Azure endpoint, API version, and auth. You can provide either an API key or a token provider. If not provided, api_version defaults to 2025-05-01-preview.

client = AsyncAzureContentUnderstandingClient(
    endpoint="https://example.cognitiveservices.azure.com",
    api_version="2025-05-01-preview",
    api_key="...",
)

Methods

• get_all_analyzers() -> Dict[str, Any]

  • Returns a dictionary with a value list of analyzers.

• get_analyzer_detail_by_id(analyzer_id: str) -> Dict[str, Any]

  • Returns details for a specific analyzer.

• begin_create_analyzer(analyzer_id: str, analyzer_schema: AnalyzerSchema) -> aiohttp.ClientResponse

  • Creates or updates an analyzer. Returns the raw HTTP response.

• delete_analyzer(analyzer_id: str) -> aiohttp.ClientResponse

  • Deletes an analyzer.

• begin_analyze_url(analyzer_id: str, url: str) -> aiohttp.ClientResponse

  • Submits an HTTP/HTTPS document URL for analysis.

• begin_analyze_binary(analyzer_id: str, file_location: str) -> aiohttp.ClientResponse

  • Submits a local file for analysis.

• poll_result(response: aiohttp.ClientResponse, timeout_seconds: int = 180, polling_interval_seconds: int = 2) -> Dict[str, Any]

  • Polls the operation until completion and returns the final payload.

• get_result_file(analyze_response: aiohttp.ClientResponse, file_id: str) -> Optional[bytes]

  • Fetches a generated result file by ID.

• begin_create_classifier(classifier_id: str, classifier_schema: ClassifierSchema) -> aiohttp.ClientResponse

  • Creates or updates a classifier.

• begin_classify(classifier_id: str, file_location: str) -> aiohttp.ClientResponse

  • Classifies a document from a local path or public URL.

Notes

• The client lazily creates an aiohttp.ClientSession if you do not provide one.
• Call await client.close() if you did not use async with.

Helper Layer (azure_cu_sdk.helpers)

ContentUnderstandingOperationsHelper

The helper wraps the async client and returns the raw service payload.

helper = ContentUnderstandingOperationsHelper(
    client=client,
    config=ContentUnderstandingExecutionConfig(analyzer_id="my-analyzer"),
)

analyze_document

payload = await helper.analyze_document(
    DocumentExecutionRequest(
        document_url="https://example.com/doc.pdf",
        file_name="doc.pdf",
        file_id="doc-001",
    ),
    analyzer_id="override-analyzer",
)

To analyze a local file, pass file_path instead of document_url:

payload = await helper.analyze_document(
    DocumentExecutionRequest(
        file_path="C:/data/doc.pdf",
        file_name="doc.pdf",
        file_id="doc-001",
    ),
    analyzer_id="override-analyzer",
)

• If analyzer_id is passed, it overrides the default in the config.
• Returns the raw Content Understanding response payload with no modifications.
• On failure, raises the underlying exception.

Analyzer Management

You can create or update analyzers using typed schemas.

from azure_cu_sdk.models import AnalyzerSchema

schema = AnalyzerSchema(
    kind="document",
    description="My analyzer",
)

resp = await helper.create_analyzer(
    analyzer_id="my-analyzer",
    analyzer_schema=schema,
)
print(resp)

DocumentExecutionRequest

• document_url: HTTP/HTTPS URL of the document.
• file_name: Original file name.
• file_id: Stable identifier used in output chunk IDs.

ContentUnderstandingExecutionConfig

• analyzer_id: Default analyzer for analysis calls.
• poll_timeout_seconds: Max time to wait for completion.
• poll_interval_seconds: Polling delay.

Raw Payload Handler

You can pass a handler to capture raw responses:

from azure_cu_sdk.helpers import default_raw_payload_logger

helper = ContentUnderstandingOperationsHelper(
    client=client,
    config=ContentUnderstandingExecutionConfig(analyzer_id="my-analyzer"),
    raw_payload_handler=default_raw_payload_logger,
)

The handler signature is:

async def handler(payload: Dict[str, Any], request: DocumentExecutionRequest) -> None: ...

Models (azure_cu_sdk.models)

Pydantic models for payload shape. These are generic and not tied to a specific analyzer schema:

• TextSource
• DocumentSource
• ImageSource
• ContentItem
• AnalyzeRequest
• AnalyzeResponse
• AnalyzerSchema
• ClassifierSchema

Exceptions (azure_cu_sdk.exceptions)

• AzureCUError: Base error
• ClientConfigurationError: Missing/invalid config
• AuthenticationError: Missing auth
• ServiceError: Non-2xx responses or service failures

Packaging

The project uses Hatchling and is ready for PyPI:

python -m build
python -m twine upload dist/*

CLI

Analyze a document URL:

azure-cu-sdk analyze --endpoint https://example.cognitiveservices.azure.com ^
  --api-key YOUR_KEY ^
  --analyzer-id my-analyzer ^
  --file-name doc.pdf ^
  --file-id doc-001 ^
  --url https://example.com/doc.pdf

Analyze a local file:

azure-cu-sdk analyze --endpoint https://example.cognitiveservices.azure.com ^
  --api-key YOUR_KEY ^
  --analyzer-id my-analyzer ^
  --file-name doc.pdf ^
  --file-id doc-001 ^
  --file C:/data/doc.pdf

Versioning

Set versions in pyproject.toml before publishing.