Skip to content

azure

State client for Azure Blob Storage.

AzureBlobStateClient

Bases: StateClient

Read and write state as JSON blobs in Azure Blob Storage.

The path argument is interpreted as the blob name inside the configured container.

Requires the azure-storage-blob package to be installed::

uv add azure-storage-blob

Authentication priority:

  1. The account_key constructor argument, when provided explicitly.
  2. The AZURE_STORAGE_ACCOUNT_KEY environment variable.

  3. DefaultAzureCredential from the azure-identity package::

    uv add azure-identity

Parameters:

Name Type Description Default
account_name str

Azure Storage Account name.

 required
container_name str

Name of the blob container to use.

 required
account_key str | None

Azure Storage Account key. When None, the AZURE_STORAGE_ACCOUNT_KEY environment variable is tried first, then DefaultAzureCredential is used as a fallback.

 None
Source code in src/cloe_delta_table_manager/state/clients/azure.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
class AzureBlobStateClient(StateClient):
    """Read and write state as JSON blobs in Azure Blob Storage.

    The ``path`` argument is interpreted as the blob name inside the configured container.

    Requires the ``azure-storage-blob`` package to be installed::

        uv add azure-storage-blob

    Authentication priority:

    1. The *account_key* constructor argument, when provided explicitly.
    2. The ``AZURE_STORAGE_ACCOUNT_KEY`` environment variable.
    3. ``DefaultAzureCredential`` from the ``azure-identity`` package::

           uv add azure-identity

    Args:
        account_name: Azure Storage Account name.
        container_name: Name of the blob container to use.
        account_key: Azure Storage Account key.  When ``None``, the
            ``AZURE_STORAGE_ACCOUNT_KEY`` environment variable is tried
            first, then ``DefaultAzureCredential`` is used as a fallback.

    """

    def __init__(
        self,
        account_name: str,
        container_name: str,
        *,
        account_key: str | None = None,
    ) -> None:
        """Initialize the Azure Blob Storage state client.

        Args:
            account_name: Azure Storage Account name.
            container_name: Name of the blob container to use.
            account_key: Azure Storage Account key.  When ``None``, the
                ``AZURE_STORAGE_ACCOUNT_KEY`` environment variable is tried
                first, then ``DefaultAzureCredential`` is used as a fallback.

        Raises:
            ImportError: If azure-storage-blob (or azure-identity when no
                account key is available) is not installed.

        """
        if BlobServiceClient is None or AzureNamedKeyCredential is None:
            raise ImportError(
                "azure-storage-blob is required for AzureBlobStateClient. Install it with: uv add azure-storage-blob"
            )
        if account_key is None:
            account_key = os.environ.get(_ACCOUNT_KEY_ENV_VAR) or None
        credential: AzureNamedKeyCredential | DefaultAzureCredential
        if account_key is not None:
            credential = AzureNamedKeyCredential(account_name, account_key)
        else:
            if DefaultAzureCredential is None:
                raise ImportError(
                    "azure-identity is required when no account_key is provided. Install it with: uv add azure-identity"
                )
            credential = DefaultAzureCredential()
        account_url = f"https://{account_name}.blob.core.windows.net"
        self._blob_service_client: BlobServiceClientType = BlobServiceClient(
            account_url=account_url, credential=credential
        )
        self._container_name = container_name

    def read_state(self, path: str, state_type: type[StateT]) -> StateT:
        """Read state from an Azure Blob Storage blob.

        Args:
            path: Blob name within the configured container.
            state_type: The concrete State subclass to deserialize into.

        Returns:
            A State instance loaded from the blob.

        Raises:
            FileNotFoundError: If the blob does not exist.

        """
        container_client = self._blob_service_client.get_container_client(self._container_name)
        blob_client = container_client.get_blob_client(path)
        if not blob_client.exists():
            raise FileNotFoundError(f"State blob not found: {self._container_name}/{path}")
        data = blob_client.download_blob().readall()
        return state_type.model_validate_json(data)

    def write_state(self, path: str, state: State) -> None:
        """Write state to an Azure Blob Storage blob.

        The container is created automatically if it does not exist.
        The version in the state metadata is incremented before writing.

        Args:
            path: Blob name within the configured container.
            state: The State instance to persist.

        """
        state.metadata.version += 1
        state.metadata.last_updated = datetime.now()
        container_client = self._blob_service_client.get_container_client(self._container_name)
        if not container_client.exists():
            container_client.create_container()
        blob_client = container_client.get_blob_client(path)
        blob_client.upload_blob(state.model_dump_json(indent=2), overwrite=True)

    def state_exists(self, path: str) -> bool:
        """Check whether a state blob exists in Azure Blob Storage.

        Args:
            path: Blob name within the configured container.

        Returns:
            True if the blob exists, False otherwise.

        """
        container_client = self._blob_service_client.get_container_client(self._container_name)
        blob_client = container_client.get_blob_client(path)
        return blob_client.exists()

__init__(account_name, container_name, *, account_key=None)

Initialize the Azure Blob Storage state client.

Parameters:

Name Type Description Default
account_name str

Azure Storage Account name.

 required
container_name str

Name of the blob container to use.

 required
account_key str | None

Azure Storage Account key. When None, the AZURE_STORAGE_ACCOUNT_KEY environment variable is tried first, then DefaultAzureCredential is used as a fallback.

 None

Raises:

Type Description
ImportError

If azure-storage-blob (or azure-identity when no account key is available) is not installed.
Source code in src/cloe_delta_table_manager/state/clients/azure.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def __init__(
    self,
    account_name: str,
    container_name: str,
    *,
    account_key: str | None = None,
) -> None:
    """Initialize the Azure Blob Storage state client.

    Args:
        account_name: Azure Storage Account name.
        container_name: Name of the blob container to use.
        account_key: Azure Storage Account key.  When ``None``, the
            ``AZURE_STORAGE_ACCOUNT_KEY`` environment variable is tried
            first, then ``DefaultAzureCredential`` is used as a fallback.

    Raises:
        ImportError: If azure-storage-blob (or azure-identity when no
            account key is available) is not installed.

    """
    if BlobServiceClient is None or AzureNamedKeyCredential is None:
        raise ImportError(
            "azure-storage-blob is required for AzureBlobStateClient. Install it with: uv add azure-storage-blob"
        )
    if account_key is None:
        account_key = os.environ.get(_ACCOUNT_KEY_ENV_VAR) or None
    credential: AzureNamedKeyCredential | DefaultAzureCredential
    if account_key is not None:
        credential = AzureNamedKeyCredential(account_name, account_key)
    else:
        if DefaultAzureCredential is None:
            raise ImportError(
                "azure-identity is required when no account_key is provided. Install it with: uv add azure-identity"
            )
        credential = DefaultAzureCredential()
    account_url = f"https://{account_name}.blob.core.windows.net"
    self._blob_service_client: BlobServiceClientType = BlobServiceClient(
        account_url=account_url, credential=credential
    )
    self._container_name = container_name

read_state(path, state_type)

Read state from an Azure Blob Storage blob.

Parameters:

Name Type Description Default
path str

Blob name within the configured container.

 required
state_type type[StateT]

The concrete State subclass to deserialize into.

 required

Returns:

Type Description
StateT

A State instance loaded from the blob.

Raises:

Type Description
FileNotFoundError

If the blob does not exist.
Source code in src/cloe_delta_table_manager/state/clients/azure.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def read_state(self, path: str, state_type: type[StateT]) -> StateT:
    """Read state from an Azure Blob Storage blob.

    Args:
        path: Blob name within the configured container.
        state_type: The concrete State subclass to deserialize into.

    Returns:
        A State instance loaded from the blob.

    Raises:
        FileNotFoundError: If the blob does not exist.

    """
    container_client = self._blob_service_client.get_container_client(self._container_name)
    blob_client = container_client.get_blob_client(path)
    if not blob_client.exists():
        raise FileNotFoundError(f"State blob not found: {self._container_name}/{path}")
    data = blob_client.download_blob().readall()
    return state_type.model_validate_json(data)

state_exists(path)

Check whether a state blob exists in Azure Blob Storage.

Parameters:

Name Type Description Default
path str

Blob name within the configured container.

 required

Returns:

Type Description
bool

True if the blob exists, False otherwise.
Source code in src/cloe_delta_table_manager/state/clients/azure.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
def state_exists(self, path: str) -> bool:
    """Check whether a state blob exists in Azure Blob Storage.

    Args:
        path: Blob name within the configured container.

    Returns:
        True if the blob exists, False otherwise.

    """
    container_client = self._blob_service_client.get_container_client(self._container_name)
    blob_client = container_client.get_blob_client(path)
    return blob_client.exists()

write_state(path, state)

Write state to an Azure Blob Storage blob.

The container is created automatically if it does not exist. The version in the state metadata is incremented before writing.

Parameters:

Name Type Description Default
path str

Blob name within the configured container.

 required
state State

The State instance to persist.

 required
Source code in src/cloe_delta_table_manager/state/clients/azure.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def write_state(self, path: str, state: State) -> None:
    """Write state to an Azure Blob Storage blob.

    The container is created automatically if it does not exist.
    The version in the state metadata is incremented before writing.

    Args:
        path: Blob name within the configured container.
        state: The State instance to persist.

    """
    state.metadata.version += 1
    state.metadata.last_updated = datetime.now()
    container_client = self._blob_service_client.get_container_client(self._container_name)
    if not container_client.exists():
        container_client.create_container()
    blob_client = container_client.get_blob_client(path)
    blob_client.upload_blob(state.model_dump_json(indent=2), overwrite=True)