azure

name	azure
description	Use when writing Python code that integrates with Azure Blob Storage, AI Search, Document Intelligence, or Key Vault — or when configuring Managed Identity auth, designing a hybrid search index, or troubleshooting Azure SDK retry behavior.

Azure SDK

Production patterns for Azure services in Python using the official Azure SDKs.

When to Activate

• Writing code that imports azure-storage-blob, azure-search-documents, azure-ai-formrecognizer, or azure-identity
• Configuring authentication for Azure services (Managed Identity, service principals, connection strings)
• Designing or querying an Azure AI Search index (vector, text, hybrid)
• Extracting content from documents using Azure Document Intelligence
• Managing secrets with Azure Key Vault
• Deploying a pipeline as an Azure Function
• Troubleshooting Azure SDK errors or retry behavior

Authentication

DefaultAzureCredential (always prefer this)

from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient

credential = DefaultAzureCredential()
client = BlobServiceClient(account_url="https://<account>.blob.core.windows.net", credential=credential)

DefaultAzureCredential tries, in order: environment variables → Managed Identity → Azure CLI → VS Code → Interactive browser. The same code works locally (via CLI auth) and in production (via Managed Identity) without changes.

# BAD: connection string hardcoded
client = BlobServiceClient.from_connection_string("DefaultEndpointsProtocol=https;AccountName=...")

# BAD: key hardcoded
client = BlobServiceClient(account_url=url, credential="storage-account-key-here")

# GOOD: keyless auth
credential = DefaultAzureCredential()
client = BlobServiceClient(account_url=url, credential=credential)

Auth decision matrix

Environment	Credential type	How to enable
Local dev	Azure CLI	az login
CI/CD	Service principal (env vars)	Set AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_TENANT_ID
Azure VM / AKS	System-assigned Managed Identity	Enable on the resource in portal/Bicep
Azure Functions	User-assigned Managed Identity	Set AZURE_CLIENT_ID env var
Testing	ClientSecretCredential	Explicit — never use in production code

Configuration (pydantic-settings)

from pydantic_settings import BaseSettings

class AzureSettings(BaseSettings):
    azure_storage_account_url: str
    azure_search_endpoint: str
    azure_search_index_name: str
    azure_document_intelligence_endpoint: str
    azure_key_vault_url: str | None = None

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

settings = AzureSettings()

Never store credentials in settings — let DefaultAzureCredential handle them.

Azure Blob Storage

Upload

from azure.storage.blob import BlobServiceClient, ContentSettings

def upload_file(account_url: str, container: str, blob_name: str, data: bytes, content_type: str) -> str:
    credential = DefaultAzureCredential()
    client = BlobServiceClient(account_url=account_url, credential=credential)
    blob = client.get_blob_client(container=container, blob=blob_name)
    blob.upload_blob(
        data,
        overwrite=True,
        content_settings=ContentSettings(content_type=content_type),
    )
    return blob.url

Download and list

def download_blob(account_url: str, container: str, blob_name: str) -> bytes:
    client = BlobServiceClient(account_url=account_url, credential=DefaultAzureCredential())
    blob = client.get_blob_client(container=container, blob=blob_name)
    return blob.download_blob().readall()

def list_blobs(account_url: str, container: str, prefix: str = "") -> list[str]:
    client = BlobServiceClient(account_url=account_url, credential=DefaultAzureCredential())
    container_client = client.get_container_client(container)
    return [b.name for b in container_client.list_blobs(name_starts_with=prefix)]

SAS token (time-limited read access)

from datetime import datetime, timedelta, timezone
from azure.storage.blob import generate_blob_sas, BlobSasPermissions

def get_sas_url(account_name: str, account_key: str, container: str, blob: str, expiry_hours: int = 1) -> str:
    sas = generate_blob_sas(
        account_name=account_name,
        container_name=container,
        blob_name=blob,
        account_key=account_key,
        permission=BlobSasPermissions(read=True),
        expiry=datetime.now(timezone.utc) + timedelta(hours=expiry_hours),
    )
    return f"https://{account_name}.blob.core.windows.net/{container}/{blob}?{sas}"

Azure AI Search

Index schema (with vector field)

from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndex, SimpleField, SearchableField, SearchFieldDataType,
    VectorSearch, HnswAlgorithmConfiguration, VectorSearchProfile,
    SearchField, SemanticConfiguration, SemanticSearch, SemanticPrioritizedFields,
    SemanticField,
)

def create_index(endpoint: str, index_name: str) -> None:
    client = SearchIndexClient(endpoint=endpoint, credential=DefaultAzureCredential())
    fields = [
        SimpleField(name="id", type=SearchFieldDataType.String, key=True),
        SearchableField(name="content", type=SearchFieldDataType.String),
        SearchableField(name="title", type=SearchFieldDataType.String),
        SimpleField(name="source", type=SearchFieldDataType.String, filterable=True),
        SimpleField(name="chunk_index", type=SearchFieldDataType.Int32, filterable=True),
        SearchField(
            name="content_vector",
            type=SearchFieldDataType.Collection(SearchFieldDataType.Single),
            searchable=True,
            vector_search_dimensions=1536,
            vector_search_profile_name="hnsw-profile",
        ),
    ]
    vector_search = VectorSearch(
        algorithms=[HnswAlgorithmConfiguration(name="hnsw")],
        profiles=[VectorSearchProfile(name="hnsw-profile", algorithm_configuration_name="hnsw")],
    )
    semantic_search = SemanticSearch(
        configurations=[
            SemanticConfiguration(
                name="default",
                prioritized_fields=SemanticPrioritizedFields(
                    content_fields=[SemanticField(field_name="content")],
                    title_field=SemanticField(field_name="title"),
                ),
            )
        ]
    )
    index = SearchIndex(
        name=index_name,
        fields=fields,
        vector_search=vector_search,
        semantic_search=semantic_search,
    )
    client.create_or_update_index(index)

Upload documents

from azure.search.documents import SearchClient

def upload_documents(endpoint: str, index_name: str, docs: list[dict]) -> None:
    client = SearchClient(
        endpoint=endpoint,
        index_name=index_name,
        credential=DefaultAzureCredential(),
    )
    # Batch in chunks of 1000 (SDK limit)
    for i in range(0, len(docs), 1000):
        result = client.upload_documents(documents=docs[i:i + 1000])
        failed = [r for r in result if not r.succeeded]
        if failed:
            raise RuntimeError(f"{len(failed)} documents failed to index: {failed[0].key}")

Search: text / vector / hybrid

from azure.search.documents.models import VectorizedQuery

def search(
    endpoint: str,
    index_name: str,
    query: str,
    query_vector: list[float],
    top: int = 5,
    mode: str = "hybrid",  # "text" | "vector" | "hybrid"
    filter_expr: str | None = None,
) -> list[dict]:
    client = SearchClient(endpoint=endpoint, index_name=index_name, credential=DefaultAzureCredential())

    vector_query = VectorizedQuery(
        vector=query_vector,
        k_nearest_neighbors=top,
        fields="content_vector",
    ) if mode in ("vector", "hybrid") else None

    results = client.search(
        search_text=query if mode in ("text", "hybrid") else None,
        vector_queries=[vector_query] if vector_query else None,
        filter=filter_expr,
        top=top,
        query_type="semantic" if mode == "hybrid" else "simple",
        semantic_configuration_name="default" if mode == "hybrid" else None,
    )
    return [dict(r) for r in results]

Search mode comparison

Mode	When to use	Relevance	Cost
Text	Keyword lookup, exact matches	Low	Lowest
Vector	Semantic similarity, paraphrase	High	Medium
Hybrid	Production RAG (default choice)	Highest	Medium
Semantic reranking	High-precision Q&A on top of hybrid	Highest	Higher

Azure Document Intelligence

from azure.ai-formrecognizer import DocumentAnalysisClient

def analyze_document(endpoint: str, file_bytes: bytes, model_id: str = "prebuilt-read") -> dict:
    client = DocumentAnalysisClient(endpoint=endpoint, credential=DefaultAzureCredential())
    poller = client.begin_analyze_document(model_id, document=file_bytes)
    result = poller.result()
    return {
        "content": result.content,
        "pages": len(result.pages),
        "tables": [
            {
                "row_count": t.row_count,
                "column_count": t.column_count,
                "cells": [{"row": c.row_index, "col": c.column_index, "text": c.content} for c in t.cells],
            }
            for t in (result.tables or [])
        ],
    }

Model selection

Model ID	Best for
prebuilt-read	Text extraction from any document
prebuilt-layout	Tables, checkboxes, structure-aware extraction
prebuilt-document	Key-value pairs + tables
prebuilt-invoice	Invoices
prebuilt-receipt	Receipts
Custom model	Domain-specific forms with consistent layout

Key Vault

from azure.keyvault.secrets import SecretClient

def get_secret(vault_url: str, secret_name: str) -> str:
    client = SecretClient(vault_url=vault_url, credential=DefaultAzureCredential())
    return client.get_secret(secret_name).value

# Cache the client — don't recreate per call
_kv_client: SecretClient | None = None

def kv_client(vault_url: str) -> SecretClient:
    global _kv_client
    if _kv_client is None:
        _kv_client = SecretClient(vault_url=vault_url, credential=DefaultAzureCredential())
    return _kv_client

Retry with tenacity

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from azure.core.exceptions import HttpResponseError, ServiceRequestError

def is_retryable(exc: Exception) -> bool:
    if isinstance(exc, HttpResponseError):
        return exc.status_code in (429, 500, 502, 503, 504)
    return isinstance(exc, ServiceRequestError)

@retry(
    retry=retry_if_exception_type((HttpResponseError, ServiceRequestError)),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    stop=stop_after_attempt(5),
    reraise=True,
)
def upload_with_retry(client: SearchClient, docs: list[dict]) -> None:
    client.upload_documents(documents=docs)

Error handling

from azure.core.exceptions import (
    HttpResponseError,
    ResourceNotFoundError,
    ResourceExistsError,
    ClientAuthenticationError,
    ServiceRequestError,
)

try:
    result = client.get_document(key="doc-123")
except ResourceNotFoundError:
    # Document does not exist — handle gracefully
    return None
except ClientAuthenticationError:
    # Credential expired or RBAC role missing — fail fast
    raise
except HttpResponseError as e:
    if e.status_code == 429:
        # Throttled — tenacity will handle retry
        raise
    logger.error("azure_error", status=e.status_code, message=e.message)
    raise

Cost controls

Lever	Impact	How
AI Search tier	High	Basic for dev, Standard S1 for prod; avoid S3 HD unless >1B docs
Semantic reranking	Medium	Enable only on queries that need it; billed per 1000 queries
Document Intelligence	Medium	Use prebuilt-read (cheapest) unless you need tables or KV pairs
Blob storage tier	Low-medium	Hot for active docs, Cool for archive; lifecycle policies auto-tier
Vector dimensions	Medium	1536 (ada-002) vs 3072 (text-embedding-3-large) — smaller = cheaper storage

Red Flags

• Hardcoded connection strings or storage account keys — keys can be leaked or rotated; always use DefaultAzureCredential with RBAC roles, never access keys or SAS tokens in code
• DefaultAzureCredential in production without pinning to ManagedIdentityCredential — the credential chain tries 6+ sources sequentially; a misconfigured chain causes 30s+ startup failures; pin to ManagedIdentityCredential in prod
• SDK clients recreated per request — SDK clients are designed to be long-lived and manage connection pools; recreating them per request exhausts connections and slows every call
• Uploading documents to AI Search one at a time — single-document uploads are ~100× slower than batching; always use upload_documents in batches of up to 1000
• Text-only search for RAG queries — semantic/vector-only search misses exact-match terms; use hybrid search (text + vector) with semantic re-ranking for best recall across diverse queries
• No retry policy on 429 or 503 responses — Azure services throttle under load; wrap all SDK calls with tenacity or the Azure SDK's built-in retry configuration
• ClientAuthenticationError silently retried — auth errors must fail fast and loudly; retrying authentication failures burns through retry budget and delays surfacing the real problem

Checklist

• All SDK clients use DefaultAzureCredential — no hardcoded keys or connection strings
• Managed Identity enabled on compute (Function App, VM, AKS node pool)
• RBAC roles assigned (Storage Blob Data Contributor, Search Index Data Contributor, etc.) — not access keys
• Secrets stored in Key Vault, not env vars or config files
• All long-running SDK calls wrapped with tenacity retry on 429/5xx
• Document upload batched in chunks of ≤1000 for Azure AI Search
• Index schema reviewed: filterable/sortable fields declared explicitly
• Hybrid search enabled for RAG queries (not text-only)
• Blob lifecycle policy configured to auto-tier cold data to Cool/Archive
• ClientAuthenticationError caught and surfaced immediately (not retried)
• SDK client instances reused per process — not recreated per request
• Azure resource names follow naming convention (<service>-<env>-<region>-<suffix>)