AWSProvider

laktory.models.resources.AWSProvider

Bases: BaseProvider, PulumiResource, TerraformResource

AWS Provider

PARAMETER	DESCRIPTION
resource_name_	Name of the resource in the context of infrastructure as code. If None, default_resource_name will be used instead. TYPE: str | VariableType DEFAULT: None
options	Resources options specifications TYPE: ResourceOptions | VariableType DEFAULT: ResourceOptions(variables={}, is_enabled=True, depends_on=[], provider=None, ignore_changes=None, aliases=None, delete_before_replace=True, import_=None, parent=None, replace_on_changes=None, moved_from=None)
lookup_existing	Lookup resource instead of creating a new one. TYPE: ResourceLookup | VariableType DEFAULT: None
variables	Dict of variables to be injected in the model at runtime TYPE: dict[str, Any] DEFAULT: {}
alias	TYPE: str | VariableType DEFAULT: None
source	TYPE: str | VariableType DEFAULT: None
version	TYPE: str | VariableType DEFAULT: None
access_key	The access key for API operations. You can retrieve this from the 'Security & Credentials' section of the AWS console. TYPE: str | VariableType DEFAULT: None
allowed_account_ids	TYPE: list[Union[str, VariableType]] | VariableType DEFAULT: None
assume_role	TYPE: ProviderAssumeRole | VariableType DEFAULT: None
assume_role_with_web_identity	TYPE: ProviderAssumeRoleWithWebIdentity | VariableType DEFAULT: None
custom_ca_bundle	File containing custom root and intermediate certificates. Can also be configured using the AWS_CA_BUNDLE environment variable. (Setting ca_bundle in the shared config file is not supported.) TYPE: str | VariableType DEFAULT: None
default_tags	Configuration block with settings to default resource tags across all resources. TYPE: ProviderDefaultTags | VariableType DEFAULT: None
ec2_metadata_service_endpoint	Address of the EC2 metadata service endpoint to use. Can also be configured using the AWS_EC2_METADATA_SERVICE_ENDPOINT environment variable. TYPE: str | VariableType DEFAULT: None
ec2_metadata_service_endpoint_mode	Protocol to use with EC2 metadata service endpoint.Valid values are IPv4 and IPv6. Can also be configured using the AWS_EC2_METADATA_SERVICE_ENDPOINT_MODE environment variable. TYPE: str | VariableType DEFAULT: None
forbidden_account_ids	TYPE: list[Union[str, VariableType]] | VariableType DEFAULT: None
http_proxy	URL of a proxy to use for HTTP requests when accessing the AWS API. Can also be set using the HTTP_PROXY or http_proxy environment variables. TYPE: str | VariableType DEFAULT: None
https_proxy	URL of a proxy to use for HTTPS requests when accessing the AWS API. Can also be set using the HTTPS_PROXY or https_proxy environment variables. TYPE: str | VariableType DEFAULT: None
ignore_tags	Configuration block with settings to ignore resource tags across all resources. TYPE: ProviderIgnoreTagsArgs | VariableType DEFAULT: None
insecure	Explicitly allow the provider to perform 'insecure' SSL requests. If omitted, default value is false TYPE: bool | VariableType DEFAULT: None
max_retries	The maximum number of times an AWS API request is being executed. If the API request still fails, an error is thrown. TYPE: int | VariableType DEFAULT: None
no_proxy	Comma-separated list of hosts that should not use HTTP or HTTPS proxies. Can also be set using the NO_PROXY or no_proxy environment variables. TYPE: str | VariableType DEFAULT: None
profile	The profile for API operations. If not set, the default profile created with aws configure will be used. TYPE: str | VariableType DEFAULT: None
region	The region where AWS operations will take place. Examples are us-east-1, us-west-2, etc. It can also be sourced from the following environment variables: AWS_REGION, AWS_DEFAULT_REGION TYPE: str | VariableType DEFAULT: None
retry_mode	Specifies how retries are attempted. Valid values are standard and adaptive. Can also be configured using the AWS_RETRY_MODE environment variable. TYPE: str | VariableType DEFAULT: None
s3_us_east1_regional_endpoint	Specifies whether S3 API calls in the us-east-1 region use the legacy global endpoint or a regional endpoint. Valid values are legacy or regional. Can also be configured using the AWS_S3_US_EAST_1_REGIONAL_ENDPOINT environment variable or the s3_us_east_1_regional_endpoint shared config file parameter TYPE: str | VariableType DEFAULT: None
s3_use_path_style	Set this to true to enable the request to use path-style addressing, i.e., https://s3.amazonaws.com/BUCKET/KEY. By default, the S3 client will use virtual hosted bucket addressing when possible (https://BUCKET.s3.amazonaws.com/KEY). Specific to the Amazon S3 service. TYPE: bool | VariableType DEFAULT: None
secret_key	The secret key for API operations. You can retrieve this from the 'Security & Credentials' section of the AWS console. TYPE: str | VariableType DEFAULT: None
shared_config_files	List of paths to shared config files. If not set, defaults to [~/.aws/config]. TYPE: list[Union[str, VariableType]] | VariableType DEFAULT: None
shared_credentials_files	List of paths to shared credentials files. If not set, defaults to [~/.aws/credentials]. TYPE: list[Union[str, VariableType]] | VariableType DEFAULT: None
skip_credentials_validation	Skip the credentials validation via STS API. Used for AWS API implementations that do not have STS available/implemented. TYPE: bool | VariableType DEFAULT: None
skip_metadata_api_check	Skip the AWS Metadata API check. Used for AWS API implementations that do not have a metadata api endpoint. TYPE: bool | VariableType DEFAULT: None
skip_region_validation	Skip static validation of region name. Used by users of alternative AWS-like APIs or users w/ access to regions that are not public (yet). TYPE: bool | VariableType DEFAULT: None
skip_requesting_account_id	Skip requesting the account ID. Used for AWS API implementations that do not have IAM/STS API and/or metadata API. TYPE: bool | VariableType DEFAULT: None
sts_region	The region where AWS STS operations will take place. Examples are us-east-1 and us-west-2. TYPE: str | VariableType DEFAULT: None
token	Session token. A session token is only required if you are using temporary security credentials. TYPE: str | VariableType DEFAULT: None
use_dualstack_endpoint	Resolve an endpoint with DualStack capability TYPE: bool | VariableType DEFAULT: None
use_fips_endpoint	Resolve an endpoint with FIPS capability TYPE: bool | VariableType DEFAULT: None

Examples:

from laktory import models

p = models.AWSProvider(
    access_key="${vars.AWS_ACCESS_KEY}",
)

METHOD	DESCRIPTION
inject_vars	Inject model variables values into a model attributes.
inject_vars_into_dump	Inject model variables values into a model dump.
model_validate_json_file	Load model from json file object
model_validate_yaml	Load model from yaml file object using laktory.yaml.RecursiveLoader. Supports
push_vars	Push variable values to all child recursively
validate_assignment_disabled	Updating a model attribute inside a model validator when validate_assignment

ATTRIBUTE	DESCRIPTION
core_resources	List of core resources to be deployed with this laktory model:
default_resource_name	Resource default name constructed as TYPE: str
pulumi_properties	Resources properties formatted for pulumi: TYPE: dict
pulumi_renames	Map of fields to rename when dumping model to pulumi TYPE: dict[str, str]
resource_key	Resource key used to build default resource name. Equivalent to TYPE: str
resource_name	dots (.) are allowed to support terraform aliases TYPE: str
resource_type_id	Resource type id used to build default resource name. Equivalent to TYPE: str
self_as_core_resources	Flag set to True if self must be included in core resources
terraform_excludes	List of fields to exclude when dumping model to terraform TYPE: Union[list[str], dict[str, bool]]
terraform_properties	Resources properties formatted for terraform: TYPE: dict
terraform_renames	Map of fields to rename when dumping model to terraform TYPE: dict[str, str]

core_resources property

List of core resources to be deployed with this laktory model: - class instance (self)

default_resource_name property

Resource default name constructed as - {self.resource_type_id}-{self.resource_key} - removing ${resources....} tags - removing ${vars....} tags - Replacing special characters with - to avoid conflicts with resource properties

pulumi_properties property

Resources properties formatted for pulumi:

• Serialization (model dump)
• Removal of excludes defined in self.pulumi_excludes
• Renaming of keys according to self.pulumi_renames
• Injection of variables

RETURNS	DESCRIPTION
dict	Pulumi-safe model dump

pulumi_renames property

Map of fields to rename when dumping model to pulumi

resource_key property

Resource key used to build default resource name. Equivalent to name properties if available. Otherwise, empty string.

resource_name property

dots (.) are allowed to support terraform aliases

resource_type_id property

Resource type id used to build default resource name. Equivalent to class name converted to kebab case. e.g.: SecretScope -> secret-scope

self_as_core_resources property

Flag set to True if self must be included in core resources

terraform_excludes property

List of fields to exclude when dumping model to terraform

terraform_properties property

Resources properties formatted for terraform:

• Serialization (model dump)
• Removal of excludes defined in self.terraform_excludes
• Renaming of keys according to self.terraform_renames
• Injection of variables

RETURNS	DESCRIPTION
dict	Terraform-safe model dump

terraform_renames property

Map of fields to rename when dumping model to terraform

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_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)

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