Skip to content

Index

State module for managing deployment and migration state.

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)

DatabaseState

Bases: State, Databases

State of the latest deployment as a collection of databases.

Source code in src/cloe_delta_table_manager/state/state.py 
64
65
66
67
68
69
class DatabaseState(State, Databases):
    """State of the latest deployment as a collection of databases."""

    # Field override required: ``Databases.databases`` has no default, so without
    # this re-declaration ``DatabaseState()`` (empty) would fail validation.
    databases: list[Database] = Field(default_factory=list)

LocalStateClient

Bases: StateClient

Read and write state as JSON files on the local filesystem.

The path argument is interpreted as a filesystem path (absolute or relative).

Source code in src/cloe_delta_table_manager/state/clients/local.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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
class LocalStateClient(StateClient):
    """Read and write state as JSON files on the local filesystem.

    The ``path`` argument is interpreted as a filesystem path (absolute or relative).
    """

    def read_state(self, path: str, state_type: type[StateT]) -> StateT:
        """Read state from a local JSON file.

        Args:
            path: Filesystem path to the JSON state file.
            state_type: The concrete State subclass to deserialize into.

        Returns:
            A State instance loaded from the file.

        Raises:
            FileNotFoundError: If the file does not exist.

        """
        file_path = Path(path)
        if not file_path.exists():
            raise FileNotFoundError(f"State file not found: {path}")
        return state_type.model_validate_json(file_path.read_text())

    def write_state(self, path: str, state: State) -> None:
        """Write state to a local JSON file.

        Parent directories are created automatically if they do not exist.
        The version in the state metadata is incremented before writing.

        Args:
            path: Filesystem path for the JSON state file.
            state: The State instance to persist.

        """
        file_path = Path(path)
        state.metadata.version += 1
        state.metadata.last_updated = datetime.now()
        file_path.parent.mkdir(parents=True, exist_ok=True)
        file_path.write_text(state.model_dump_json(indent=2))

    def state_exists(self, path: str) -> bool:
        """Check whether a state file exists on the local filesystem.

        Args:
            path: Filesystem path to check.

        Returns:
            True if the file exists, False otherwise.

        """
        return Path(path).exists()

read_state(path, state_type)

Read state from a local JSON file.

Parameters:

Name Type Description Default
path str

Filesystem path to the JSON state file.

 required
state_type type[StateT]

The concrete State subclass to deserialize into.

 required

Returns:

Type Description
StateT

A State instance loaded from the file.

Raises:

Type Description
FileNotFoundError

If the file does not exist.
Source code in src/cloe_delta_table_manager/state/clients/local.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def read_state(self, path: str, state_type: type[StateT]) -> StateT:
    """Read state from a local JSON file.

    Args:
        path: Filesystem path to the JSON state file.
        state_type: The concrete State subclass to deserialize into.

    Returns:
        A State instance loaded from the file.

    Raises:
        FileNotFoundError: If the file does not exist.

    """
    file_path = Path(path)
    if not file_path.exists():
        raise FileNotFoundError(f"State file not found: {path}")
    return state_type.model_validate_json(file_path.read_text())

state_exists(path)

Check whether a state file exists on the local filesystem.

Parameters:

Name Type Description Default
path str

Filesystem path to check.

 required

Returns:

Type Description
bool

True if the file exists, False otherwise.
Source code in src/cloe_delta_table_manager/state/clients/local.py 
55
56
57
58
59
60
61
62
63
64
65
def state_exists(self, path: str) -> bool:
    """Check whether a state file exists on the local filesystem.

    Args:
        path: Filesystem path to check.

    Returns:
        True if the file exists, False otherwise.

    """
    return Path(path).exists()

write_state(path, state)

Write state to a local JSON file.

Parent directories are created automatically if they do not exist. The version in the state metadata is incremented before writing.

Parameters:

Name Type Description Default
path str

Filesystem path for the JSON state file.

 required
state State

The State instance to persist.

 required
Source code in src/cloe_delta_table_manager/state/clients/local.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def write_state(self, path: str, state: State) -> None:
    """Write state to a local JSON file.

    Parent directories are created automatically if they do not exist.
    The version in the state metadata is incremented before writing.

    Args:
        path: Filesystem path for the JSON state file.
        state: The State instance to persist.

    """
    file_path = Path(path)
    state.metadata.version += 1
    state.metadata.last_updated = datetime.now()
    file_path.parent.mkdir(parents=True, exist_ok=True)
    file_path.write_text(state.model_dump_json(indent=2))

MigrationState

Bases: State

State of the migration.

Source code in src/cloe_delta_table_manager/state/state.py 
72
73
74
75
class MigrationState(State):
    """State of the migration."""

    migrations: list[Migration | FileMigration] = []

State

Bases: BaseModel

State of the latest deployment.

Source code in src/cloe_delta_table_manager/state/state.py 
28
29
30
31
32
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
class State(BaseModel):
    """State of the latest deployment."""

    metadata: StateMetadata = StateMetadata()

    def to_file(self, path: str | Path, client: StateClient | None = None) -> None:
        """Save state using the given client, defaulting to the local filesystem.

        Args:
            path: Location identifier for the state (file path for local, blob
                name for Azure, etc.).
            client: State client to use.  Defaults to ``LocalStateClient``.

        """
        from cloe_delta_table_manager.state.clients.local import LocalStateClient  # noqa: PLC0415

        (client or LocalStateClient()).write_state(str(path), self)

    @classmethod
    def from_file(cls, path: str | Path, client: StateClient | None = None) -> Self:
        """Load state using the given client, defaulting to the local filesystem.

        Args:
            path: Location identifier for the state (file path for local, blob
                name for Azure, etc.).
            client: State client to use.  Defaults to ``LocalStateClient``.

        Returns:
            State instance loaded from the given path.

        """
        from cloe_delta_table_manager.state.clients.local import LocalStateClient  # noqa: PLC0415

        return (client or LocalStateClient()).read_state(str(path), cls)

from_file(path, client=None) classmethod

Load state using the given client, defaulting to the local filesystem.

Parameters:

Name Type Description Default
path str | Path

Location identifier for the state (file path for local, blob name for Azure, etc.).

 required
client StateClient | None

State client to use. Defaults to LocalStateClient.

 None

Returns:

Type Description
Self

State instance loaded from the given path.
Source code in src/cloe_delta_table_manager/state/state.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@classmethod
def from_file(cls, path: str | Path, client: StateClient | None = None) -> Self:
    """Load state using the given client, defaulting to the local filesystem.

    Args:
        path: Location identifier for the state (file path for local, blob
            name for Azure, etc.).
        client: State client to use.  Defaults to ``LocalStateClient``.

    Returns:
        State instance loaded from the given path.

    """
    from cloe_delta_table_manager.state.clients.local import LocalStateClient  # noqa: PLC0415

    return (client or LocalStateClient()).read_state(str(path), cls)

to_file(path, client=None)

Save state using the given client, defaulting to the local filesystem.

Parameters:

Name Type Description Default
path str | Path

Location identifier for the state (file path for local, blob name for Azure, etc.).

 required
client StateClient | None

State client to use. Defaults to LocalStateClient.

 None
Source code in src/cloe_delta_table_manager/state/state.py 
33
34
35
36
37
38
39
40
41
42
43
44
def to_file(self, path: str | Path, client: StateClient | None = None) -> None:
    """Save state using the given client, defaulting to the local filesystem.

    Args:
        path: Location identifier for the state (file path for local, blob
            name for Azure, etc.).
        client: State client to use.  Defaults to ``LocalStateClient``.

    """
    from cloe_delta_table_manager.state.clients.local import LocalStateClient  # noqa: PLC0415

    (client or LocalStateClient()).write_state(str(path), self)

StateClient

Bases: ABC

Abstract base class for reading and writing state from different source systems.

Source code in src/cloe_delta_table_manager/state/clients/base.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class StateClient(ABC):
    """Abstract base class for reading and writing state from different source systems."""

    @abstractmethod
    def read_state(self, path: str, state_type: type[StateT]) -> StateT:
        """Read state from the source system.

        Args:
            path: Location identifier for the state (interpretation depends on the client).
            state_type: The concrete State subclass to deserialize into.

        Returns:
            A State instance loaded from the source system.

        Raises:
            FileNotFoundError: If the state does not exist at the given path.

        """

    @abstractmethod
    def write_state(self, path: str, state: State) -> None:
        """Write state to the source system.

        The version in the state metadata is incremented before writing.

        Args:
            path: Location identifier for the state (interpretation depends on the client).
            state: The State instance to persist.

        """

    @abstractmethod
    def state_exists(self, path: str) -> bool:
        """Check whether a state exists at the given path.

        Args:
            path: Location identifier for the state.

        Returns:
            True if the state exists, False otherwise.

        """

read_state(path, state_type) abstractmethod

Read state from the source system.

Parameters:

Name Type Description Default
path str

Location identifier for the state (interpretation depends on the client).

 required
state_type type[StateT]

The concrete State subclass to deserialize into.

 required

Returns:

Type Description
StateT

A State instance loaded from the source system.

Raises:

Type Description
FileNotFoundError

If the state does not exist at the given path.
Source code in src/cloe_delta_table_manager/state/clients/base.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@abstractmethod
def read_state(self, path: str, state_type: type[StateT]) -> StateT:
    """Read state from the source system.

    Args:
        path: Location identifier for the state (interpretation depends on the client).
        state_type: The concrete State subclass to deserialize into.

    Returns:
        A State instance loaded from the source system.

    Raises:
        FileNotFoundError: If the state does not exist at the given path.

    """

state_exists(path) abstractmethod

Check whether a state exists at the given path.

Parameters:

Name Type Description Default
path str

Location identifier for the state.

 required

Returns:

Type Description
bool

True if the state exists, False otherwise.
Source code in src/cloe_delta_table_manager/state/clients/base.py 
42
43
44
45
46
47
48
49
50
51
52
@abstractmethod
def state_exists(self, path: str) -> bool:
    """Check whether a state exists at the given path.

    Args:
        path: Location identifier for the state.

    Returns:
        True if the state exists, False otherwise.

    """

write_state(path, state) abstractmethod

Write state to the source system.

The version in the state metadata is incremented before writing.

Parameters:

Name Type Description Default
path str

Location identifier for the state (interpretation depends on the client).

 required
state State

The State instance to persist.

 required
Source code in src/cloe_delta_table_manager/state/clients/base.py 
30
31
32
33
34
35
36
37
38
39
40
@abstractmethod
def write_state(self, path: str, state: State) -> None:
    """Write state to the source system.

    The version in the state metadata is incremented before writing.

    Args:
        path: Location identifier for the state (interpretation depends on the client).
        state: The State instance to persist.

    """

StateMetadata

Bases: BaseModel

Metadata of the state.

Source code in src/cloe_delta_table_manager/state/state.py 
20
21
22
23
24
25
class StateMetadata(BaseModel):
    """Metadata of the state."""

    version: int = 0
    last_updated: datetime = Field(default_factory=datetime.now)
    delta_table_manager_version: str = lib_version("cloe-delta-table-manager")