Skip to content

TerraformStack

laktory.models.TerraformStack ¤

Bases: BaseModel

A Terraform stack is terraform-specific flavor of the laktory.models.Stack. It re-structure the attributes to be aligned with a terraform json file.

It is generally not instantiated directly, but rather created using laktory.models.Stack.to_terraform().

PARAMETER DESCRIPTION
variables

Dict of variables to be injected in the model at runtime

TYPE: dict[str, Any] DEFAULT: {}

terraform

TYPE: TerraformConfig | VariableType DEFAULT: TerraformConfig(variables={}, required_providers=None, backend=None)

providers

TYPE: dict[Union[str, VariableType], Union[Any, VariableType]] | VariableType DEFAULT: {}

resources

TYPE: dict[Union[str, VariableType], Union[Any, VariableType]] | VariableType DEFAULT: {}

References
METHOD DESCRIPTION
apply

Runs terraform apply

destroy

Runs terraform destroy

init

Runs terraform init

inject_vars

Inject model variables values into a model attributes.

inject_vars_into_dump

Inject model variables values into a model dump.

model_dump

Serialize model to match the structure of a Terraform json file.

model_validate_json_file

Load model from json file object

model_validate_yaml

Load model from yaml file object using laktory.yaml.RecursiveLoader. Supports

plan

Runs terraform plan

push_vars

Push variable values to all child recursively

validate_assignment_disabled

Updating a model attribute inside a model validator when validate_assignment

write

Write Terraform json configuration file

apply(flags=None) ¤

Runs terraform apply

PARAMETER DESCRIPTION
flags

List of flags / options for terraform apply

TYPE: list[str] DEFAULT: None

Source code in laktory/models/stacks/terraformstack.py
241
242
243
244
245
246
247
248
249
250
def apply(self, flags: list[str] = None):
    """
    Runs `terraform apply`

    Parameters
    ----------
    flags:
        List of flags / options for terraform apply
    """
    self._call("apply", flags=flags)

destroy(flags=None) ¤

Runs terraform destroy

PARAMETER DESCRIPTION
flags

List of flags / options for terraform destroy

TYPE: list[str] DEFAULT: None

Source code in laktory/models/stacks/terraformstack.py
252
253
254
255
256
257
258
259
260
261
def destroy(self, flags: list[str] = None):
    """
    Runs `terraform destroy`

    Parameters
    ----------
    flags:
        List of flags / options for terraform destroy
    """
    self._call("destroy", flags=flags)

init(flags=None) ¤

Runs terraform init

PARAMETER DESCRIPTION
flags

List of flags / options for pulumi plan

TYPE: list[str] DEFAULT: None

Source code in laktory/models/stacks/terraformstack.py
219
220
221
222
223
224
225
226
227
228
def init(self, flags: list[str] = None) -> None:
    """
    Runs `terraform init`

    Parameters
    ----------
    flags:
        List of flags / options for pulumi plan
    """
    self._call("init", flags=flags)

inject_vars(inplace=False, vars=None) ¤

Inject model variables values into a model attributes.

PARAMETER DESCRIPTION
inplace

If True model is modified in place. Otherwise, a new model instance is returned.

TYPE: bool DEFAULT: False

vars

A dictionary of variables to be injected in addition to the model internal variables.

TYPE: dict DEFAULT: None

RETURNS DESCRIPTION

Model instance.

Examples:

from typing import Union

from laktory import models


class Cluster(models.BaseModel):
    name: str = None
    size: Union[int, str] = None


c = Cluster(
    name="cluster-${vars.my_cluster}",
    size="${{ 4 if vars.env == 'prod' else 2 }}",
    variables={
        "env": "dev",
    },
).inject_vars()
print(c)
# > variables={'env': 'dev'} name='cluster-${vars.my_cluster}' size=2
References
Source code in laktory/models/basemodel.py
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
def inject_vars(self, inplace: bool = False, vars: dict = None):
    """
    Inject model variables values into a model attributes.

    Parameters
    ----------
    inplace:
        If `True` model is modified in place. Otherwise, a new model
        instance is returned.
    vars:
        A dictionary of variables to be injected in addition to the
        model internal variables.


    Returns
    -------
    :
        Model instance.

    Examples
    --------
    ```py
    from typing import Union

    from laktory import models


    class Cluster(models.BaseModel):
        name: str = None
        size: Union[int, str] = None


    c = Cluster(
        name="cluster-${vars.my_cluster}",
        size="${{ 4 if vars.env == 'prod' else 2 }}",
        variables={
            "env": "dev",
        },
    ).inject_vars()
    print(c)
    # > variables={'env': 'dev'} name='cluster-${vars.my_cluster}' size=2
    ```

    References
    ----------
    * [variables](https://www.laktory.ai/concepts/variables/)
    """

    # Fetching vars
    if vars is None:
        vars = {}
    vars = deepcopy(vars)
    vars.update(self.variables)

    # Create copy
    if not inplace:
        self = self.model_copy(deep=True)

    # Inject into field values
    for k in list(self.model_fields_set):
        if k == "variables":
            continue
        o = getattr(self, k)

        if isinstance(o, BaseModel) or isinstance(o, dict) or isinstance(o, list):
            # Mutable objects will be updated in place
            _resolve_values(o, vars)
        else:
            # Simple objects must be updated explicitly
            setattr(self, k, _resolve_value(o, vars))

    # Inject into child resources
    if hasattr(self, "core_resources"):
        for r in self.core_resources:
            if r == self:
                continue
            r.inject_vars(vars=vars, inplace=True)

    if not inplace:
        return self

inject_vars_into_dump(dump, inplace=False, vars=None) ¤

Inject model variables values into a model dump.

PARAMETER DESCRIPTION
dump

Model dump (or any other general purpose mutable object)

TYPE: dict[str, Any]

inplace

If True model is modified in place. Otherwise, a new model instance is returned.

TYPE: bool DEFAULT: False

vars

A dictionary of variables to be injected in addition to the model internal variables.

TYPE: dict[str, Any] DEFAULT: None

RETURNS DESCRIPTION

Model dump with injected variables.

Examples:

from laktory import models

m = models.BaseModel(
    variables={
        "env": "dev",
    },
)
data = {
    "name": "cluster-${vars.my_cluster}",
    "size": "${{ 4 if vars.env == 'prod' else 2 }}",
}
print(m.inject_vars_into_dump(data))
# > {'name': 'cluster-${vars.my_cluster}', 'size': 2}
References
Source code in laktory/models/basemodel.py
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
def inject_vars_into_dump(
    self, dump: dict[str, Any], inplace: bool = False, vars: dict[str, Any] = None
):
    """
    Inject model variables values into a model dump.

    Parameters
    ----------
    dump:
        Model dump (or any other general purpose mutable object)
    inplace:
        If `True` model is modified in place. Otherwise, a new model
        instance is returned.
    vars:
        A dictionary of variables to be injected in addition to the
        model internal variables.


    Returns
    -------
    :
        Model dump with injected variables.


    Examples
    --------
    ```py
    from laktory import models

    m = models.BaseModel(
        variables={
            "env": "dev",
        },
    )
    data = {
        "name": "cluster-${vars.my_cluster}",
        "size": "${{ 4 if vars.env == 'prod' else 2 }}",
    }
    print(m.inject_vars_into_dump(data))
    # > {'name': 'cluster-${vars.my_cluster}', 'size': 2}
    ```

    References
    ----------
    * [variables](https://www.laktory.ai/concepts/variables/)
    """

    # Setting vars
    if vars is None:
        vars = {}
    vars = deepcopy(vars)
    vars.update(self.variables)

    # Create copy
    if not inplace:
        dump = copy.deepcopy(dump)

    # Inject into field values
    _resolve_values(dump, vars)

    if not inplace:
        return dump

model_dump(*args, **kwargs) ¤

Serialize model to match the structure of a Terraform json file.

Source code in laktory/models/stacks/terraformstack.py
 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
154
155
def model_dump(self, *args, **kwargs) -> dict[str, Any]:
    """Serialize model to match the structure of a Terraform json file."""
    self._configure_serializer(singular=True)
    kwargs["exclude_none"] = kwargs.get("exclude_none", True)
    d = super().model_dump(*args, **kwargs)

    # Special treatment of resources
    d["resource"] = defaultdict(lambda: {})
    d["data"] = defaultdict(lambda: {})
    for r in self.resources.values():
        _d = r.terraform_properties
        if r.lookup_existing:
            d["data"][r.terraform_resource_lookup_type][r.resource_name] = (
                r.lookup_existing.model_dump()
            )
            for k in r.options.terraform_options:
                if k in _d:
                    d["data"][r.terraform_resource_lookup_type][r.resource_name][
                        k
                    ] = _d[k]
        else:
            d["resource"][r.terraform_resource_type][r.resource_name] = _d
    d["data"] = dict(d["data"])
    if len(d["data"]) == 0:
        del d["data"]
    d["resource"] = dict(d["resource"])
    self._configure_serializer(singular=False)

    # Special treatment of moved
    # `moved_from` should generally be used with Terraform, but we also
    # support aliases (Pulumi) for better user experience
    i = -1
    for r in self.resources.values():
        moved_from = r.options.moved_from
        aliases = r.options.aliases
        _from = None
        if moved_from:
            _from = moved_from
        elif aliases:
            _from = aliases[0]
        if _from:
            i += 1
            d[f"moved_{i:05d}"] = {
                "from": f"{r.terraform_resource_type}.{_from}",
                "to": f"{r.terraform_resource_type}.{r.resource_name}",
            }

    # Terraform JSON requires the keyword "resources." to be removed and the
    # resource_name to be replaced with resource_type.resource_name.
    _vars = {}
    for r in list(self.resources.values()) + list(self.providers.values()):
        k0 = r.resource_name

        k1 = f"{r.terraform_resource_type}.{r.resource_name}"
        # special treatment for data sources
        if r.lookup_existing:
            k1 = f"data.{r.terraform_resource_lookup_type}.{r.resource_name}"

        if isinstance(r, BaseProvider):
            pattern = r"\$\{resources." + k0 + "}"
            _vars[pattern] = k0

        else:
            # ${resources.resource_name} -> resource_type.resource_name
            pattern = r"\$\{resources\." + k0 + r"\}"
            _vars[pattern] = k1

            # ${resources.resource_name.property} -> ${resource_type.resource_name.property}
            pattern = r"\$\{resources\." + k0 + r"\.(.*?)\}"
            _vars[pattern] = rf"${{{k1}.\1}}"

    # Because all variables are mapped to a string, it is more efficient
    # (>10x) to convert the dict to string before substitution.
    d = json.loads(_resolve_values(json.dumps(d), vars=_vars))

    return d

model_validate_json_file(fp) classmethod ¤

Load model from json file object

PARAMETER DESCRIPTION
fp

file object structured as a json file

TYPE: TextIO

RETURNS DESCRIPTION
Model

Model instance

Source code in laktory/models/basemodel.py
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
@classmethod
def model_validate_json_file(cls: Type[Model], fp: TextIO) -> Model:
    """
    Load model from json file object

    Parameters
    ----------
    fp:
        file object structured as a json file

    Returns
    -------
    :
        Model instance
    """
    data = json.load(fp)
    return cls.model_validate(data)

model_validate_yaml(fp) classmethod ¤

Load model from yaml file object using laktory.yaml.RecursiveLoader. Supports reference to external yaml and sql files using !use, !extend and !update tags. Path to external files can be defined using model or environment variables.

Referenced path should always be relative to the file they are referenced from.

Custom Tags
  • !use {filepath}: Directly inject the content of the file at filepath

  • - !extend {filepath}: Extend the current list with the elements found in the file at filepath. Similar to python list.extend method.

  • <<: !update {filepath}: Merge the current dictionary with the content of the dictionary defined at filepath. Similar to python dict.update method.

PARAMETER DESCRIPTION
fp

file object structured as a yaml file

TYPE: TextIO

RETURNS DESCRIPTION
Model

Model instance

Examples:

businesses:
  apple:
    symbol: aapl
    address: !use addresses.yaml
    <<: !update common.yaml
    emails:
      - jane.doe@apple.com
      - extend! emails.yaml
  amazon:
    symbol: amzn
    address: !use addresses.yaml
    <<: update! common.yaml
    emails:
      - john.doe@amazon.com
      - extend! emails.yaml
Source code in laktory/models/basemodel.py
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
@classmethod
def model_validate_yaml(cls: Type[Model], fp: TextIO) -> Model:
    """
    Load model from yaml file object using laktory.yaml.RecursiveLoader. Supports
    reference to external yaml and sql files using `!use`, `!extend` and `!update` tags.
    Path to external files can be defined using model or environment variables.

    Referenced path should always be relative to the file they are referenced from.

    Custom Tags
    -----------
    - `!use {filepath}`:
        Directly inject the content of the file at `filepath`

    - `- !extend {filepath}`:
        Extend the current list with the elements found in the file at `filepath`.
        Similar to python list.extend method.

    - `<<: !update {filepath}`:
        Merge the current dictionary with the content of the dictionary defined at
        `filepath`. Similar to python dict.update method.

    Parameters
    ----------
    fp:
        file object structured as a yaml file

    Returns
    -------
    :
        Model instance

    Examples
    --------
    ```yaml
    businesses:
      apple:
        symbol: aapl
        address: !use addresses.yaml
        <<: !update common.yaml
        emails:
          - jane.doe@apple.com
          - extend! emails.yaml
      amazon:
        symbol: amzn
        address: !use addresses.yaml
        <<: update! common.yaml
        emails:
          - john.doe@amazon.com
          - extend! emails.yaml
    ```
    """

    data = RecursiveLoader.load(fp)
    return cls.model_validate(data)

plan(flags=None) ¤

Runs terraform plan

PARAMETER DESCRIPTION
flags

List of flags / options for pulumi plan

TYPE: list[str] DEFAULT: None

Source code in laktory/models/stacks/terraformstack.py
230
231
232
233
234
235
236
237
238
239
def plan(self, flags: list[str] = None) -> None:
    """
    Runs `terraform plan`

    Parameters
    ----------
    flags:
        List of flags / options for pulumi plan
    """
    self._call("plan", flags=flags)

push_vars(update_core_resources=False) ¤

Push variable values to all child recursively

Source code in laktory/models/basemodel.py
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
def push_vars(self, update_core_resources=False) -> Any:
    """Push variable values to all child recursively"""

    def _update_model(m):
        if not isinstance(m, BaseModel):
            return
        for k, v in self.variables.items():
            m.variables[k] = m.variables.get(k, v)
        m.push_vars()

    def _push_vars(o):
        if isinstance(o, list):
            for _o in o:
                _push_vars(_o)
        elif isinstance(o, dict):
            for _o in o.values():
                _push_vars(_o)
        else:
            _update_model(o)

    for k in self.model_fields.keys():
        _push_vars(getattr(self, k))

    if update_core_resources and hasattr(self, "core_resources"):
        for r in self.core_resources:
            if r != self:
                _push_vars(r)

    return None

validate_assignment_disabled() ¤

Updating a model attribute inside a model validator when validate_assignment is True causes an infinite recursion by design and must be turned off temporarily.

Source code in laktory/models/basemodel.py
323
324
325
326
327
328
329
330
331
332
333
334
335
@contextmanager
def validate_assignment_disabled(self):
    """
    Updating a model attribute inside a model validator when `validate_assignment`
    is `True` causes an infinite recursion by design and must be turned off
    temporarily.
    """
    original_state = self.model_config["validate_assignment"]
    self.model_config["validate_assignment"] = False
    try:
        yield
    finally:
        self.model_config["validate_assignment"] = original_state

write() ¤

Write Terraform json configuration file

RETURNS DESCRIPTION
str

Filepath of the configuration file

Source code in laktory/models/stacks/terraformstack.py
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def write(self) -> str:
    """
    Write Terraform json configuration file

    Returns
    -------
    :
        Filepath of the configuration file
    """
    filepath = os.path.join(CACHE_ROOT, "stack.tf.json")
    logger.info(f"Writing terraform config at '{filepath}'")

    if not os.path.exists(CACHE_ROOT):
        os.makedirs(CACHE_ROOT)

    text = json.dumps(self.model_dump(), indent=4)

    # Terraform stack file is not a strict format. Some keys might be
    # repeated and require special treatment.

    # Special treatment of providers with aliases
    for _, p in self.providers.items():
        if p.alias is not None:
            text = text.replace(
                f'"{p.resource_name}":',
                f'"{p.resource_name_without_alias}":',
            )

    # Special treatment of moved
    text = re.sub(r'"moved_\d+": {', '"moved": {', text)

    # Output
    with open(filepath, "w") as fp:
        fp.write(text)

    return filepath