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

• Stack

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

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

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

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

• variables

Source code in laktory/models/basemodel.py

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

• variables

Source code in laktory/models/basemodel.py

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

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

@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

@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

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

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

@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

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