Data persistence and serialization in Durable Functions for Azure Functions

This example requires the Azure.Storage.Blobs NuGet package.

using System;
using System.Threading.Tasks;
using Azure.Storage.Blobs;
using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;

public record BlobReference(string Container, string BlobName);

public static class LargePayloadFunctions
{
    [Function("ProcessLargePayload")]
    public static async Task<BlobReference> RunOrchestrator(
        [OrchestrationTrigger] TaskOrchestrationContext context)
    {
        BlobReference inputReference = context.GetInput<BlobReference>()
            ?? throw new InvalidOperationException("A blob reference is required.");

        return await context.CallActivityAsync<BlobReference>(
            nameof(ProcessLargePayloadActivity), inputReference);
    }

    [Function(nameof(ProcessLargePayloadActivity))]
    public static async Task<BlobReference> ProcessLargePayloadActivity(
        [ActivityTrigger] BlobReference inputReference)
    {
        string connectionString =
            Environment.GetEnvironmentVariable("PAYLOAD_STORAGE_CONNECTION_STRING")
            ?? throw new InvalidOperationException("Payload storage is not configured.");
        BlobServiceClient service = new BlobServiceClient(connectionString);

        BlobClient inputBlob = service
            .GetBlobContainerClient(inputReference.Container)
            .GetBlobClient(inputReference.BlobName);
        BinaryData inputData = (await inputBlob.DownloadContentAsync()).Value.Content;

        BlobContainerClient outputContainer =
            service.GetBlobContainerClient("processed-payloads");
        await outputContainer.CreateIfNotExistsAsync();

        string outputName = $"processed/{Guid.NewGuid():N}.json";
        await outputContainer.GetBlobClient(outputName)
            .UploadAsync(inputData, overwrite: true);

        return new BlobReference(outputContainer.Name, outputName);
    }
}

This example requires the Azure.Storage.Blobs NuGet package.

using System;
using System.Threading.Tasks;
using Azure.Storage.Blobs;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.DurableTask;

public class BlobReference
{
    public string Container { get; set; } = string.Empty;
    public string BlobName { get; set; } = string.Empty;
}

public static class LargePayloadFunctions
{
    [FunctionName("ProcessLargePayload")]
    public static async Task<BlobReference> RunOrchestrator(
        [OrchestrationTrigger] IDurableOrchestrationContext context)
    {
        BlobReference inputReference = context.GetInput<BlobReference>()
            ?? throw new InvalidOperationException("A blob reference is required.");

        return await context.CallActivityAsync<BlobReference>(
            "ProcessLargePayloadActivity", inputReference);
    }

    [FunctionName("ProcessLargePayloadActivity")]
    public static async Task<BlobReference> RunActivity(
        [ActivityTrigger] BlobReference inputReference)
    {
        string connectionString =
            Environment.GetEnvironmentVariable("PAYLOAD_STORAGE_CONNECTION_STRING")
            ?? throw new InvalidOperationException("Payload storage is not configured.");
        BlobServiceClient service = new BlobServiceClient(connectionString);

        BlobClient inputBlob = service
            .GetBlobContainerClient(inputReference.Container)
            .GetBlobClient(inputReference.BlobName);
        BinaryData inputData = (await inputBlob.DownloadContentAsync()).Value.Content;

        BlobContainerClient outputContainer =
            service.GetBlobContainerClient("processed-payloads");
        await outputContainer.CreateIfNotExistsAsync();

        string outputName = $"processed/{Guid.NewGuid():N}.json";
        await outputContainer.GetBlobClient(outputName)
            .UploadAsync(inputData, overwrite: true);

        return new BlobReference
        {
            Container = outputContainer.Name,
            BlobName = outputName,
        };
    }
}

This example requires the @azure/storage-blob npm package.

const { randomUUID } = require("crypto");
const { BlobServiceClient } = require("@azure/storage-blob");
const df = require("durable-functions");

df.app.orchestration("processLargePayload", function* (context) {
    const inputReference = context.df.getInput();
    return yield context.df.callActivity(
        "processLargePayloadActivity", inputReference);
});

df.app.activity("processLargePayloadActivity", {
    handler: async (inputReference) => {
        const connectionString =
            process.env.PAYLOAD_STORAGE_CONNECTION_STRING;
        if (!connectionString) {
            throw new Error(
                "PAYLOAD_STORAGE_CONNECTION_STRING is not set.");
        }

        const service = BlobServiceClient.fromConnectionString(
            connectionString);

        const inputBlob = service
            .getContainerClient(inputReference.container)
            .getBlockBlobClient(inputReference.blobName);
        const inputData = await inputBlob.downloadToBuffer();

        const outputContainer =
            service.getContainerClient("processed-payloads");
        await outputContainer.createIfNotExists();

        const outputName = `processed/${randomUUID()}.json`;
        await outputContainer.getBlockBlobClient(outputName)
            .uploadData(inputData);

        return {
            container: outputContainer.containerName,
            blobName: outputName,
        };
    },
});

This example requires the azure-storage-blob package. Add it to requirements.txt.

import os
import uuid

import azure.functions as func
import azure.durable_functions as df
from azure.core.exceptions import ResourceExistsError
from azure.storage.blob import BlobServiceClient

app = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)


@app.orchestration_trigger(context_name="context")
def process_large_payload(context: df.DurableOrchestrationContext):
    input_reference: dict = context.get_input()
    return (yield context.call_activity(
        "process_large_payload_activity", input_reference))


@app.activity_trigger(input_name="input_reference")
def process_large_payload_activity(input_reference: dict) -> dict:
    connection_string = os.getenv("PAYLOAD_STORAGE_CONNECTION_STRING")
    if not connection_string:
        raise ValueError(
            "PAYLOAD_STORAGE_CONNECTION_STRING is not set.")

    service = BlobServiceClient.from_connection_string(connection_string)

    input_blob = service.get_blob_client(
        container=input_reference["container"],
        blob=input_reference["blobName"])
    input_data = input_blob.download_blob().readall()

    output_container = service.get_container_client("processed-payloads")
    try:
        output_container.create_container()
    except ResourceExistsError:
        pass

    output_name = f"processed/{uuid.uuid4()}.json"
    output_container.upload_blob(
        name=output_name, data=input_data, overwrite=True)

    return {
        "container": output_container.container_name,
        "blobName": output_name,
    }

This example requires the Az.Storage module. Add it to requirements.psd1 and make sure managed dependencies are enabled in host.json.

Orchestrator run.ps1:

param($Context)

$OutputReference = Invoke-DurableActivity `
    -FunctionName 'ProcessLargePayloadActivity' `
    -Input $Context.Input

return $OutputReference

Activity run.ps1:

param($InputReference)

if ([string]::IsNullOrEmpty(
        $env:PAYLOAD_STORAGE_CONNECTION_STRING)) {
    throw 'The PAYLOAD_STORAGE_CONNECTION_STRING app setting is not configured.'
}

$StorageContext = New-AzStorageContext `
    -ConnectionString $env:PAYLOAD_STORAGE_CONNECTION_STRING
$InputFile = New-TemporaryFile

try {
    Get-AzStorageBlobContent `
        -Container $InputReference.container `
        -Blob $InputReference.blobName `
        -Destination $InputFile.FullName `
        -Context $StorageContext `
        -Force | Out-Null

    # Process $InputFile here. This example uploads it unchanged.
    $OutputContainer = 'processed-payloads'
    New-AzStorageContainer `
        -Name $OutputContainer `
        -Context $StorageContext `
        -ErrorAction SilentlyContinue | Out-Null

    $OutputName = "processed/$([guid]::NewGuid()).json"
    Set-AzStorageBlobContent `
        -File $InputFile.FullName `
        -Container $OutputContainer `
        -Blob $OutputName `
        -Context $StorageContext `
        -Force | Out-Null

    return @{
        container = $OutputContainer
        blobName = $OutputName
    }
}
finally {
    Remove-Item $InputFile.FullName -Force
}

Use the standard orchestrationTrigger and activityTrigger bindings described in Durable Functions bindings.

This example requires the com.azure:azure-storage-blob and com.microsoft:durabletask-azure-functions Maven packages.

import com.azure.core.util.BinaryData;
import com.azure.storage.blob.BlobContainerClient;
import com.azure.storage.blob.BlobServiceClient;
import com.azure.storage.blob.BlobServiceClientBuilder;
import com.microsoft.azure.functions.annotation.FunctionName;
import com.microsoft.durabletask.TaskOrchestrationContext;
import com.microsoft.durabletask.azurefunctions.DurableActivityTrigger;
import com.microsoft.durabletask.azurefunctions.DurableOrchestrationTrigger;

import java.util.Objects;
import java.util.UUID;

public class LargePayloadFunctions {
    public static class BlobReference {
        public String container;
        public String blobName;

        public BlobReference() {
        }

        public BlobReference(String container, String blobName) {
            this.container = container;
            this.blobName = blobName;
        }
    }

    @FunctionName("ProcessLargePayload")
    public BlobReference runOrchestrator(
            @DurableOrchestrationTrigger(name = "ctx")
            TaskOrchestrationContext ctx) {
        BlobReference inputReference = Objects.requireNonNull(
            ctx.getInput(BlobReference.class),
            "A blob reference is required.");
        return ctx.callActivity(
            "ProcessLargePayloadActivity",
            inputReference,
            BlobReference.class).await();
    }

    @FunctionName("ProcessLargePayloadActivity")
    public BlobReference runActivity(
            @DurableActivityTrigger(name = "inputReference")
            BlobReference inputReference) {
        String connectionString = Objects.requireNonNull(
            System.getenv("PAYLOAD_STORAGE_CONNECTION_STRING"),
            "PAYLOAD_STORAGE_CONNECTION_STRING is not configured.");
        BlobServiceClient service = new BlobServiceClientBuilder()
            .connectionString(connectionString)
            .buildClient();

        BinaryData inputData = service
            .getBlobContainerClient(inputReference.container)
            .getBlobClient(inputReference.blobName)
            .downloadContent();

        BlobContainerClient outputContainer =
            service.getBlobContainerClient("processed-payloads");
        outputContainer.createIfNotExists();

        String outputName =
            "processed/" + UUID.randomUUID() + ".json";
        outputContainer.getBlobClient(outputName)
            .upload(inputData, true);

        return new BlobReference(
            outputContainer.getBlobContainerName(), outputName);
    }
}

.NET Isolated and System.Text.Json

Durable Functions running in the .NET Isolated worker process use the same object serializer configured globally for your Azure Functions app (see WorkerOptions). This serializer is System.Text.Json by default rather than Newtonsoft.Json. Any changes to WorkerOptions.Serializer transitively apply to Durable Functions.

For more information on the built-in support for JSON serialization in .NET, see the JSON serialization and deserialization in .NET overview documentation.

Default serialization logic

Durable Functions for .NET in-process internally uses Json.NET to serialize orchestration and entity data to JSON. The default Json.NET settings used are:

Inputs, Outputs, and State:

JsonSerializerSettings
{
    TypeNameHandling = TypeNameHandling.None,
    DateParseHandling = DateParseHandling.None,
}

Exceptions:

JsonSerializerSettings
{
    ContractResolver = new ExceptionResolver(),
    TypeNameHandling = TypeNameHandling.Objects,
    ReferenceLoopHandling = ReferenceLoopHandling.Ignore,
}

Read more detailed documentation about JsonSerializerSettings here.

Customize serialization with .NET attributes

During serialization, Json.NET looks for various attributes on classes and properties that control how the data is serialized and deserialized from JSON. If you own the source code for data type passed to Durable Functions APIs, consider adding these attributes to the type to customize serialization and deserialization.

Customize serialization with Dependency Injection

Function apps that target .NET and run on the Functions V3 runtime can use Dependency Injection (DI) to customize how data and exceptions are serialized. The following sample code demonstrates how to use DI to override the default Json.NET serialization settings using custom implementations of the IMessageSerializerSettingsFactory and IErrorSerializerSettingsFactory service interfaces.

using Microsoft.Azure.Functions.Extensions.DependencyInjection;
using Microsoft.Azure.WebJobs.Extensions.DurableTask;
using Microsoft.Extensions.DependencyInjection;
using Newtonsoft.Json;
using System.Collections.Generic;

[assembly: FunctionsStartup(typeof(MyApplication.Startup))]
namespace MyApplication
{
    public class Startup : FunctionsStartup
    {
        public override void Configure(IFunctionsHostBuilder builder)
        {
            builder.Services.AddSingleton<IMessageSerializerSettingsFactory, CustomMessageSerializerSettingsFactory>();
            builder.Services.AddSingleton<IErrorSerializerSettingsFactory, CustomErrorSerializerSettingsFactory>();
        }

        /// <summary>
        /// A factory that provides the serialization for all inputs and outputs for activities and
        /// orchestrations, as well as entity state.
        /// </summary>
        internal class CustomMessageSerializerSettingsFactory : IMessageSerializerSettingsFactory
        {
            public JsonSerializerSettings CreateJsonSerializerSettings()
            {
                // Return your custom JsonSerializerSettings here
            }
        }

        /// <summary>
        /// A factory that provides the serialization for all exceptions thrown by activities
        /// and orchestrations
        /// </summary>
        internal class CustomErrorSerializerSettingsFactory : IErrorSerializerSettingsFactory
        {
            public JsonSerializerSettings CreateJsonSerializerSettings()
            {
                // Return your custom JsonSerializerSettings here
            }
        }
    }
}

Serialization and deserialization logic

Azure Functions Node applications use JSON.stringify() for serialization and JSON.Parse() for deserialization. Most types should serialize and deserialize seamlessly. In cases where the default logic is insufficient, defining a toJSON() method on the object will override the serialization logic. However, no analog exists for object deserialization.

For full customization of the serialization/deserialization pipeline, consider handling the serialization and deserialization with your own code and passing around data as strings.

Serialization and deserialization logic

Durable Functions PowerShell apps use the default PowerShell JSON handling for orchestration and activity payloads. In most cases, plain JSON-compatible objects serialize and deserialize correctly without extra configuration.

If you need custom behavior for a payload shape, serialize or deserialize the data yourself inside your functions and pass the resulting JSON-compatible values between orchestrators, activities, and entities. This approach keeps the data format explicit and avoids depending on a serializer customization API that isn't exposed for PowerShell.

For more complex scenarios, consider representing your data as simple strings, arrays, or hashtables and converting between formats inside your function code.

Serialization and deserialization logic

It's recommended to use type annotations to ensure Durable Functions serializes and deserializes your data correctly. While many built-in types are handled automatically, some built-in data types require type annotations to preserve the type during deserialization.

For custom data types, you make JSON serialization and deserialization possible by defining class methods to_json and from_json on your data type class. Note that these methods are not called on the return value from the orchestrator function, meaning the return value has to be natively JSON-serializable. For more information, see Bindings.

To validate deserialized payloads against an expected type and optionally opt in to a hardened strict mode, see Migrate to type-safe serialization in Durable Functions for Python.

Java uses the Jackson v2.x libraries for serialization and deserialization of data payloads. You can use Jackson annotations on your POJO types to customize the serialization behavior.