Skip to content

Storage Provider Types

Storage provider is one of the building blocks in Terraflex -
it defines an abstract way to communicate with a storage of any kind.
A new storage provider can easily be provided by implementing one of the protocols below - and registering into terraflex.plugins.storage_provider entrypoint.

Readable Storage

StorageProviderProtocol

Protocol for storage providers - Read only.

Readable storage is the most basic storage provider - it allows to read files from the storage.

Every readonly storage provider must implement StorageProviderProtocol methods - and register to the terraflex.plugins.storage_provider entrypoint.

Source code in terraflex/server/storage_provider_base.py 
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
@runtime_checkable
class StorageProviderProtocol(Protocol):
    """Protocol for storage providers - Read only.

    Readable storage is the most basic storage provider - it allows to read files from the storage.

    Every readonly storage provider must implement `StorageProviderProtocol` methods -
    and register to the `terraflex.plugins.storage_provider` entrypoint.
    """

    @classmethod
    async def from_config(
        cls,
        raw_config: Any,
        *,
        manager: DependenciesManager,
        workdir: pathlib.Path,
    ) -> Self:
        """Create an instance of the storage provider from the configuration.

        Args:
            raw_config: The raw configuration propagated from the storage provider config.
            manager: The dependencies manager - allows to request a binary path from.
            workdir: The data directory of terraflex - located at `~/.local/share/terraflex` -
                can be used to manage state of the provider
        """
        ...

    @classmethod
    def validate_key(cls, key: dict[str, Any]) -> ItemKey:
        """Validate the key of the item in the storage provider from a config of usage.

        Args:
            key: a dict with parameters to build the key of the item.

        Returns:
            The validated key.
        """
        ...

    # read related
    async def get_file(self, item_identifier: ItemKey) -> bytes:
        """Get the content of the file.

        Args:
            item_identifier: The identifier of the file.
        """
        ...

from_config(raw_config, *, manager, workdir) async classmethod

Create an instance of the storage provider from the configuration.

Parameters:

Name Type Description Default
raw_config Any

The raw configuration propagated from the storage provider config.

 required
manager DependenciesManager

The dependencies manager - allows to request a binary path from.

 required
workdir Path

The data directory of terraflex - located at ~/.local/share/terraflex - can be used to manage state of the provider

 required
Source code in terraflex/server/storage_provider_base.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@classmethod
async def from_config(
    cls,
    raw_config: Any,
    *,
    manager: DependenciesManager,
    workdir: pathlib.Path,
) -> Self:
    """Create an instance of the storage provider from the configuration.

    Args:
        raw_config: The raw configuration propagated from the storage provider config.
        manager: The dependencies manager - allows to request a binary path from.
        workdir: The data directory of terraflex - located at `~/.local/share/terraflex` -
            can be used to manage state of the provider
    """
    ...

get_file(item_identifier) async

Get the content of the file.

Parameters:

Name Type Description Default
item_identifier ItemKey

The identifier of the file.

 required
Source code in terraflex/server/storage_provider_base.py 
76
77
78
79
80
81
82
async def get_file(self, item_identifier: ItemKey) -> bytes:
    """Get the content of the file.

    Args:
        item_identifier: The identifier of the file.
    """
    ...

validate_key(key) classmethod

Validate the key of the item in the storage provider from a config of usage.

Parameters:

Name Type Description Default
key dict[str, Any]

a dict with parameters to build the key of the item.

 required

Returns:

Type Description
ItemKey

The validated key.
Source code in terraflex/server/storage_provider_base.py 
63
64
65
66
67
68
69
70
71
72
73
@classmethod
def validate_key(cls, key: dict[str, Any]) -> ItemKey:
    """Validate the key of the item in the storage provider from a config of usage.

    Args:
        key: a dict with parameters to build the key of the item.

    Returns:
        The validated key.
    """
    ...

ItemKey

Bases: BaseModel, ABC

Params required to reference an item in a storage provider.

Every storage provider must implement a subclass of ItemKey to represent the key of the item - using the validate_key() method.

Source code in terraflex/server/storage_provider_base.py 
13
14
15
16
17
18
19
20
21
22
class ItemKey(BaseModel, abc.ABC):
    """Params required to reference an item in a storage provider.

    Every storage provider must implement a subclass of `ItemKey` to represent the key of the item - using the validate_key() method.
    """

    @abc.abstractmethod
    def as_string(self) -> str:
        """Return the string representation of the key."""
        ...

as_string() abstractmethod

Return the string representation of the key.

Source code in terraflex/server/storage_provider_base.py 
19
20
21
22
@abc.abstractmethod
def as_string(self) -> str:
    """Return the string representation of the key."""
    ...

Writeable Storage

WriteableStorageProviderProtocol

Bases: StorageProviderProtocol, Protocol

Protocol for storage providers - Writeable.

Writeable storage provider allows to write files to the storage.

Every writeable storage provider must implement WriteableStorageProviderProtocol methods and its parent methods - and register to the terraflex.plugins.storage_provider entrypoint.

Source code in terraflex/server/storage_provider_base.py 
 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
@runtime_checkable
class WriteableStorageProviderProtocol(StorageProviderProtocol, Protocol):
    """Protocol for storage providers - Writeable.

    Writeable storage provider allows to write files to the storage.

    Every writeable storage provider must implement `WriteableStorageProviderProtocol` methods and its parent methods -
        and register to the `terraflex.plugins.storage_provider` entrypoint.
    """

    # write related
    async def put_file(self, item_identifier: ItemKey, data: bytes) -> None:
        """Put the content of the file in the provided file.

        Args:
            item_identifier: The identifier of the file.
            data: The content of the file.
        """
        ...

    async def delete_file(self, item_identifier: ItemKey) -> None:
        """Delete the file.

        Args:
            item_identifier: The identifier of the file.
        """
        ...

delete_file(item_identifier) async

Delete the file.

Parameters:

Name Type Description Default
item_identifier ItemKey

The identifier of the file.

 required
Source code in terraflex/server/storage_provider_base.py 
105
106
107
108
109
110
111
async def delete_file(self, item_identifier: ItemKey) -> None:
    """Delete the file.

    Args:
        item_identifier: The identifier of the file.
    """
    ...

put_file(item_identifier, data) async

Put the content of the file in the provided file.

Parameters:

Name Type Description Default
item_identifier ItemKey

The identifier of the file.

 required
data bytes

The content of the file.

 required
Source code in terraflex/server/storage_provider_base.py 
 96
 97
 98
 99
100
101
102
103
async def put_file(self, item_identifier: ItemKey, data: bytes) -> None:
    """Put the content of the file in the provided file.

    Args:
        item_identifier: The identifier of the file.
        data: The content of the file.
    """
    ...

Lockable Storage

LockableStorageProviderProtocol

Bases: WriteableStorageProviderProtocol, Protocol

Protocol for storage providers - Lockable.

Lockable storage provider allows to lock items to prevent concurrent writes. Allows to support terraform state locking - see more at official docs.

Every lockable storage provider must implement LockableStorageProviderProtocol methods and its parent methods - and register to the terraflex.plugins.storage_provider entrypoint.

Source code in terraflex/server/storage_provider_base.py 
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
@runtime_checkable
class LockableStorageProviderProtocol(WriteableStorageProviderProtocol, Protocol):
    """Protocol for storage providers - Lockable.

    Lockable storage provider allows to lock items to prevent concurrent writes.
    Allows to support terraform state locking - see more at official [docs](https://developer.hashicorp.com/terraform/language/state/locking).

    Every lockable storage provider must implement `LockableStorageProviderProtocol` methods and its parent methods -
        and register to the `terraflex.plugins.storage_provider` entrypoint.
    """

    # lock related
    async def read_lock(self, item_identifier: ItemKey) -> LockBody:
        """Read the lock of the item.

        Args:
            item_identifier: The identifier of the item.

        Returns:
            The lock data of the item.
        """
        ...

    async def acquire_lock(self, item_identifier: ItemKey, data: LockBody) -> None:
        """Acquire the lock of the item.

        Args:
            item_identifier: The identifier of the item.
            data: The lock data of the item.
        """
        ...

    async def release_lock(self, item_identifier: ItemKey) -> None: ...

acquire_lock(item_identifier, data) async

Acquire the lock of the item.

Parameters:

Name Type Description Default
item_identifier ItemKey

The identifier of the item.

 required
data LockBody

The lock data of the item.

 required
Source code in terraflex/server/storage_provider_base.py 
137
138
139
140
141
142
143
144
async def acquire_lock(self, item_identifier: ItemKey, data: LockBody) -> None:
    """Acquire the lock of the item.

    Args:
        item_identifier: The identifier of the item.
        data: The lock data of the item.
    """
    ...

read_lock(item_identifier) async

Read the lock of the item.

Parameters:

Name Type Description Default
item_identifier ItemKey

The identifier of the item.

 required

Returns:

Type Description
LockBody

The lock data of the item.
Source code in terraflex/server/storage_provider_base.py 
126
127
128
129
130
131
132
133
134
135
async def read_lock(self, item_identifier: ItemKey) -> LockBody:
    """Read the lock of the item.

    Args:
        item_identifier: The identifier of the item.

    Returns:
        The lock data of the item.
    """
    ...

release_lock(item_identifier) async

Source code in terraflex/server/storage_provider_base.py 
146
async def release_lock(self, item_identifier: ItemKey) -> None: ...

LockBody

Bases: BaseModel

Data struct that contains the lock information.

This is the same data struct that is required by terraform. It follows the same fields names as the terraform lock file.

See offical source.

Attributes:

Name Type Description
ID str

The ID of the lock.
Operation str

The operation that is being performed.
Who str

The entity that is performing the operation.
Version str

The version of the lock.
Created str

The time when the lock was created.
Source code in terraflex/server/base_state_lock_provider.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class LockBody(BaseModel):
    """Data struct that contains the lock information.

    This is the same data struct that is required by terraform.
    It follows the same fields names as the terraform lock file.

    See offical [source](https://github.com/hashicorp/terraform/blob/aea5c0cc180e0e6915454b3bf61f471c230c111b/internal/states/statemgr/locker.go#L129).

    Attributes:
        ID: The ID of the lock.
        Operation: The operation that is being performed.
        Who: The entity that is performing the operation.
        Version: The version of the lock.
        Created: The time when the lock was created.
    """

    model_config = ConfigDict(from_attributes=True)

    ID: str
    Operation: str
    Who: str
    Version: str
    Created: str