Skip to content

Blob Storage

This recipe covers integrating Azure Blob Storage with Azure Functions Python v2 — using output bindings to upload blobs, input bindings to read blobs, and the SDK approach for more complex scenarios like listing, streaming, and managing containers.

Architecture

flowchart TD
    BLOB[Blob Storage Event] --> TRIG[Blob Trigger]
    TRIG --> FA[Function App]
    FA --> DOWN[Downstream Service]

Solid arrows show runtime data/event flow. Dashed arrows show identity and authentication.

Prerequisites

Blob Storage bindings are included in the default extension bundle. Ensure your host.json has:

{
  "version": "2.0",
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
  }
}

You need a Storage Account with a blob container:

# Create a storage account (if not using the one provisioned with your function app)
az storage account create \
  --name yourstorage \
  --resource-group your-rg \
  --sku Standard_LRS

# Create a blob container
az storage container create \
  --name uploads \
  --account-name yourstorage
CLI element Explanation
Command(s) az storage account create, az storage container create
Key flags --name, --resource-group, --sku, --account-name
Variables None
Expected result Azure CLI returns provisioning details; confirm the resource name and successful provisioning state before continuing.

On classic Consumption or Premium plans, the AzureWebJobsStorage connection string (already set for Azure Functions) can be reused, or you can configure a separate connection. On Flex Consumption, host storage uses identity-based settings (e.g. AzureWebJobsStorage__accountName); bindings referencing connection="AzureWebJobsStorage" resolve through those identity-based settings automatically.

Output Binding: Upload Blob from HTTP Request

Use the Blob output binding to write content to Blob Storage when an HTTP request arrives:

Note

The connection parameter names a setting prefix, not a literal connection string. On Flex Consumption, AzureWebJobsStorage resolves through identity-based settings such as AzureWebJobsStorage__accountName.

import azure.functions as func
import json
import uuid

bp = func.Blueprint()

@bp.route(route="files/upload", methods=["POST"])
@bp.blob_output(
    arg_name="outputBlob",
    path="uploads/{rand-guid}.json",
    connection="AzureWebJobsStorage"
)
def upload_file(req: func.HttpRequest, outputBlob: func.Out[str]) -> func.HttpResponse:
    """Upload JSON data to Blob Storage via HTTP POST."""
    try:
        body = req.get_json()
    except ValueError:
        return func.HttpResponse(
            json.dumps({"error": "Invalid JSON body"}),
            mimetype="application/json",
            status_code=400
        )

    # Write the body to blob storage
    blob_content = json.dumps(body, indent=2)
    outputBlob.set(blob_content)

    return func.HttpResponse(
        json.dumps({"status": "uploaded", "message": "File stored in Blob Storage"}),
        mimetype="application/json",
        status_code=201
    )

The {rand-guid} binding expression generates a unique blob name for each invocation. You can also use a route parameter:

@bp.route(route="files/{filename}", methods=["PUT"])
@bp.blob_output(
    arg_name="outputBlob",
    path="uploads/{filename}",
    connection="AzureWebJobsStorage"
)
def upload_named_file(req: func.HttpRequest, outputBlob: func.Out[bytes]) -> func.HttpResponse:
    """Upload a file with a specific name."""
    content = req.get_body()
    outputBlob.set(content)

    filename = req.route_params.get("filename")
    return func.HttpResponse(
        json.dumps({"status": "uploaded", "filename": filename}),
        mimetype="application/json",
        status_code=201
    )

Input Binding: Read Blob from HTTP Request

Use the Blob input binding to read a blob when an HTTP request arrives:

@bp.route(route="files/{filename}", methods=["GET"])
@bp.blob_input(
    arg_name="inputBlob",
    path="uploads/{filename}",
    connection="AzureWebJobsStorage"
)
def download_file(req: func.HttpRequest, inputBlob: func.InputStream) -> func.HttpResponse:
    """Download a file from Blob Storage by name."""
    if inputBlob is None:
        return func.HttpResponse(
            json.dumps({"error": "File not found"}),
            mimetype="application/json",
            status_code=404
        )

    content = inputBlob.read()
    return func.HttpResponse(
        content,
        mimetype="application/octet-stream",
        status_code=200,
        headers={"Content-Disposition": f"attachment; filename={inputBlob.name}"}
    )

Note: The {filename} route parameter is automatically resolved in the binding path expression.

Blob Trigger

The Blob trigger fires when a new or updated blob is detected in a container. It is ideal for file processing pipelines — image resizing, CSV ingestion, file validation, and data transformation.

Blueprint

# blueprints/blob_processor.py
import azure.functions as func
import logging

bp = func.Blueprint()
logger = logging.getLogger(__name__)


@bp.blob_trigger(
    arg_name="input_blob",
    path="uploads/{name}",
    connection="AzureWebJobsStorage",
)
@bp.blob_output(
    arg_name="output_blob",
    path="processed/{name}",
    connection="AzureWebJobsStorage",
)
def process_blob(input_blob: func.InputStream, output_blob: func.Out[bytes]) -> None:
    """Triggered when a blob is uploaded to 'uploads'; writes result to 'processed'."""
    logger.info(
        "Processing blob: name=%s, size=%d bytes",
        input_blob.name,
        input_blob.length,
    )
    content = input_blob.read()
    output_blob.set(content.upper())
    blob_name = input_blob.name.removeprefix("uploads/")
    logger.info("Written to processed/%s", blob_name)

Register the blueprint in function_app.py:

from blueprints.blob_processor import bp as blob_processor_bp
app.register_blueprint(blob_processor_bp)

Test Locally with Azurite

Blob triggers require Azurite (the local Azure Storage emulator). Ensure Azurite is running before starting the functions host.

1. Start Azurite

azurite --silent --location /tmp/azurite --debug /tmp/azurite/debug.log

2. Create the required containers

az storage container create \
  --name uploads \
  --connection-string "UseDevelopmentStorage=true"

az storage container create \
  --name processed \
  --connection-string "UseDevelopmentStorage=true"
CLI element Explanation
Command(s) az storage container create
Key flags --name, --connection-string
Variables None
Expected result Azure CLI returns provisioning details; confirm the resource name and successful provisioning state before continuing.

3. Start the Functions host

cd app
func host start

4. Upload a test blob

echo "hello world" > /tmp/test.txt

az storage blob upload \
  --container-name uploads \
  --name test.txt \
  --file /tmp/test.txt \
  --connection-string "UseDevelopmentStorage=true"
CLI element Explanation
Command(s) az storage blob upload
Key flags --container-name, --name, --file, --connection-string
Variables None
Expected result Azure CLI completes successfully and returns JSON, table, or no output depending on the command; verify the next documented check before continuing.

The trigger is detected after a short polling delay (typically a few seconds locally). You should see in the terminal:

Processing blob: name=uploads/test.txt, size=12 bytes
Written to processed/test.txt

5. Verify the output blob

az storage blob download \
  --container-name processed \
  --name test.txt \
  --file /tmp/result.txt \
  --connection-string "UseDevelopmentStorage=true"

cat /tmp/result.txt   # Expected: HELLO WORLD
CLI element Explanation
Command(s) az storage blob download
Key flags --container-name, --name, --file, --connection-string
Variables None
Expected result Azure CLI completes successfully and returns JSON, table, or no output depending on the command; verify the next documented check before continuing.

Polling vs. Event Grid: The default blob trigger uses a polling mechanism, which can have delays of up to several minutes in production depending on storage activity. For low-latency, event-driven processing use the Event Grid-based blob trigger (Microsoft Learn) instead.

Flex Consumption: The default polling blob trigger is not supported on the Flex Consumption plan. Flex Consumption requires the Event Grid-based blob trigger. See Supported triggers on Flex Consumption (Microsoft Learn) for details.

SDK Approach: azure-storage-blob

For HTTP-triggered scenarios that need more control — listing blobs, deleting blobs, setting metadata, generating SAS tokens — use the azure-storage-blob SDK directly.

Add to requirements.txt:

azure-storage-blob>=12.19.0
azure-identity>=1.15.0

List Blobs

import azure.functions as func
import json
import os
from azure.storage.blob import BlobServiceClient
from azure.identity import DefaultAzureCredential

bp = func.Blueprint()

def get_blob_service_client() -> BlobServiceClient:
    account_name = os.environ.get("STORAGE_ACCOUNT_NAME", "yourstorage")
    credential = DefaultAzureCredential()
    return BlobServiceClient(
        account_url=f"https://{account_name}.blob.core.windows.net",
        credential=credential
    )


@bp.route(route="files", methods=["GET"])
def list_files(req: func.HttpRequest) -> func.HttpResponse:
    """List all files in the uploads container."""
    client = get_blob_service_client()
    container = client.get_container_client("uploads")

    blobs = []
    for blob in container.list_blobs():
        blobs.append({
            "name": blob.name,
            "size": blob.size,
            "last_modified": blob.last_modified.isoformat() if blob.last_modified else None,
            "content_type": blob.content_settings.content_type
        })

    return func.HttpResponse(
        json.dumps({"files": blobs, "count": len(blobs)}),
        mimetype="application/json",
        status_code=200
    )

Delete a Blob

@bp.route(route="files/{filename}", methods=["DELETE"])
def delete_file(req: func.HttpRequest) -> func.HttpResponse:
    """Delete a file from Blob Storage."""
    filename = req.route_params.get("filename")
    client = get_blob_service_client()
    blob_client = client.get_blob_client("uploads", filename)

    try:
        blob_client.delete_blob()
        return func.HttpResponse(status_code=204)
    except Exception:
        return func.HttpResponse(
            json.dumps({"error": f"File '{filename}' not found"}),
            mimetype="application/json",
            status_code=404
        )

Managed Identity for Passwordless Access

Use Managed Identity instead of connection strings to access Blob Storage:

  1. Enable identity and assign a role: 
    az functionapp identity assign --name your-func --resource-group your-rg

az role assignment create \
   --assignee "<object-id>" \
   --role "Storage Blob Data Contributor" \
   --scope "/subscriptions/<subscription-id>/resourceGroups/your-rg/providers/Microsoft.Storage/storageAccounts/yourstorage"
CLI element Explanation
Command(s) az functionapp identity assign, az role assignment create
Key flags --name, --resource-group, --assignee, --role, --scope
Variables None
Expected result Azure CLI returns provisioning details; confirm the resource name and successful provisioning state before continuing.
  1. For bindings, use identity-based connection: 
    az functionapp config appsettings set \
  --name your-func \
  --resource-group your-rg \
  --settings "AzureWebJobsStorage__accountName=yourstorage"
CLI element Explanation
Command(s) az functionapp config appsettings set
Key flags --name, --resource-group, --settings
Variables None
Expected result Azure CLI applies the configuration change; confirm the returned JSON or follow-up query shows the expected value.

See the Managed Identity recipe for a full walkthrough.

See Also

Sources