TerraformStack

laktory.models.stacks.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().

References

• Stack

PARAMETER	DESCRIPTION
providers	TYPE: dict[str | VariableType, Any | VariableType] | VariableType DEFAULT: {}
resources	TYPE: dict[str | VariableType, Any | VariableType] | VariableType DEFAULT: {}
terraform	TYPE: TerraformConfig | VariableType DEFAULT: TerraformConfig(variables={}, required_providers=None, backend=None)

METHOD	DESCRIPTION
apply	Runs terraform apply
destroy	Runs terraform destroy
init	Runs terraform init
model_dump	Serialize model to match the structure of a Terraform json file.
plan	Runs terraform plan
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 terraform 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 terraform plan
    """
    self._call("init", flags=flags)

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."""
    kwargs["exclude_none"] = kwargs.get("exclude_none", True)
    d = super().model_dump(*args, **kwargs)

    # Terraform uses singular top-level block names
    d["provider"] = d.pop("providers", {})
    d.pop("resources", None)

    # 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.resource_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"])

    # Special treatment of moved blocks
    i = -1
    for r in self.resources.values():
        _from = r.resource_options.moved_from
        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}",
            }

    # Special treatment of import
    i = -1
    for r in self.resources.values():
        import_ = r.resource_options.import_
        if import_:
            i += 1
            d[f"import_{i:05d}"] = {
                "id": import_,
                "to": f"{r.terraform_resource_type}.{r.resource_name}",
            }

    # Translate ${resources.NAME} and ${resources.NAME.PROP} references into
    # Terraform cross-references using a single combined regex so the dict is
    # scanned once, not once per resource (O(N) vs the previous O(N×M)).
    resource_lookup = {
        r.resource_name: r
        for r in list(self.resources.values()) + list(self.providers.values())
    }
    if resource_lookup:
        # Sort longest-first so that e.g. "databricks.dev" is tried before
        # "databricks" - otherwise the shorter name wins in the alternation
        # and ".dev" is misread as a property reference.
        names_alt = "|".join(
            re.escape(n) for n in sorted(resource_lookup, key=len, reverse=True)
        )
        _pattern = re.compile(
            r"\$\{resources\.(" + names_alt + r")(?:\.([^}]*))?\}"
        )

        def _replacer(m, _lookup=resource_lookup):
            name, prop = m.group(1), m.group(2)
            r = _lookup[name]
            if isinstance(r, BaseProvider):
                base = name
            elif r.lookup_existing:
                base = f"data.{r.terraform_resource_lookup_type}.{name}"
            else:
                base = f"{r.terraform_resource_type}.{name}"
            return f"${{{base}.{prop}}}" if prop is not None else base

        d = _substitute_terraform_refs(d, _pattern, _replacer)

    return d

plan(flags=None)

Runs terraform plan

PARAMETER	DESCRIPTION
flags	List of flags / options for terraform 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 terraform plan
    """
    self._call("plan", flags=flags)

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)

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

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

    return filepath