Skip to content

Reference

This part of the project documentation focuses on an information-oriented approach. Use it as a reference for the technical implementation of the aws_terraform_registry project code.

ApplicationConfig dataclass

Define aws terraform private registry parameters.

Attributes:

secret_key_name (str): AWS Secret manager name where JWT Secret is stored
repository_url (str): HTTPS endpoint of the registry
dynamodb_table_name (str): dynamodb table name
bucket_name (str): bucket name
default_namespace: default namespace to publish terrafor module ("devops" per default)
Source code in aws_terraform_registry/config.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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
83
@envclass
@dataclass
class ApplicationConfig:
    """Define aws terraform private registry parameters.


    Attributes:

        secret_key_name (str): AWS Secret manager name where JWT Secret is stored
        repository_url (str): HTTPS endpoint of the registry
        dynamodb_table_name (str): dynamodb table name
        bucket_name (str): bucket name
        default_namespace: default namespace to publish terrafor module ("devops" per default)
    """

    secret_key_name: Optional[str] = None
    repository_url: Optional[str] = None
    dynamodb_table_name: Optional[str] = None
    bucket_name: Optional[str] = None
    default_namespace: str = "devops"

    def __post_init__(self):
        """Finalize configuration.

        Feed attributs from TFR_xxxx env variable if exists.
        """
        # Feed from env var
        load_env(self, prefix='tfr')

        # remove ending '/'
        if self.repository_url and self.repository_url.endswith('/'):
            self.repository_url = self.repository_url[0:-2]

    def validate(self):
        """Validate each attributs.

        Raise:
            (RuntimeError): if an attribut is empty

        """
        for name in ['secret_key_name', 'repository_url', 'dynamodb_table_name', 'bucket_name', 'default_namespace']:
            if not getattr(self, name):
                logger.error(f"Configuration ERROR: '{name}' parameter is missing")
                raise RuntimeError(f"Configuration ERROR: '{name}' parameter is missing")

    def show(self):
        yaml.safe_dump(asdict(self), sys.stdout)

    @classmethod
    def lookup(cls, config_file_name: Optional[str] = None) -> 'ApplicationConfig':
        # define config file name
        config_file_name = config_file_name if config_file_name else os.environ.get('TFR_CONFIG_FILE', _CONFIG_NAME)

        # lookup
        for path in [Path.home(), Path.cwd(), Path('/etc/tfr')]:
            config_path = path / config_file_name
            if config_path.exists():
                return ApplicationConfig.load_from(filename=config_path)

        return ApplicationConfig()

    @classmethod
    def load_from(cls, filename: Union[Path, str]) -> 'ApplicationConfig':
        """Load ApplicationConfig from a yaml file."""
        with open(filename) as f:
            return cls(**yaml.safe_load(f.read()))

__post_init__()

Finalize configuration.

Feed attributs from TFR_xxxx env variable if exists.

Source code in aws_terraform_registry/config.py 
39
40
41
42
43
44
45
46
47
48
49
def __post_init__(self):
    """Finalize configuration.

    Feed attributs from TFR_xxxx env variable if exists.
    """
    # Feed from env var
    load_env(self, prefix='tfr')

    # remove ending '/'
    if self.repository_url and self.repository_url.endswith('/'):
        self.repository_url = self.repository_url[0:-2]

load_from(filename) classmethod

Load ApplicationConfig from a yaml file.

Source code in aws_terraform_registry/config.py 
79
80
81
82
83
@classmethod
def load_from(cls, filename: Union[Path, str]) -> 'ApplicationConfig':
    """Load ApplicationConfig from a yaml file."""
    with open(filename) as f:
        return cls(**yaml.safe_load(f.read()))

validate()

Validate each attributs.

Raise

(RuntimeError): if an attribut is empty

Source code in aws_terraform_registry/config.py 
51
52
53
54
55
56
57
58
59
60
61
def validate(self):
    """Validate each attributs.

    Raise:
        (RuntimeError): if an attribut is empty

    """
    for name in ['secret_key_name', 'repository_url', 'dynamodb_table_name', 'bucket_name', 'default_namespace']:
        if not getattr(self, name):
            logger.error(f"Configuration ERROR: '{name}' parameter is missing")
            raise RuntimeError(f"Configuration ERROR: '{name}' parameter is missing")

TerraformModuleIdentifier dataclass

Define a Terraform Module Identifier.

Attributes:

namespace (str): is the name of a namespace, unique on a particular hostname,
    that can contain one or more modules that are somehow related.
name (str):  the module name
system (str): the name of a remote system that the module is primarily written to target,
    like aws or azurerm
Source code in aws_terraform_registry/common/model.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@dataclass()
class TerraformModuleIdentifier:
    """Define a Terraform Module Identifier.

    Attributes:

        namespace (str): is the name of a namespace, unique on a particular hostname,
            that can contain one or more modules that are somehow related.
        name (str):  the module name
        system (str): the name of a remote system that the module is primarily written to target,
            like aws or azurerm
    """

    namespace: str
    name: str
    system: str

    @property
    def module_id(self) -> str:
        return f"{self.namespace}/{self.name}/{self.system}".lower()

    def get_bucket_key(self, version: str) -> str:
        """Return bucket key."""
        return f"{self.module_id}/{version}/{BUCKET_FILE_NAME}"

    def get_publish_url(self, bucket_name: str, version: str) -> str:
        """Return s3 url."""

        region = _find_caller_region()
        bucket_sub_name = f"s3-{region}" if region != "us-east-1" else "s3"
        return "/".join(
            [f"s3::https://{bucket_name}.{bucket_sub_name}.amazonaws.com", self.get_bucket_key(version=version)]
        )

    def get_blob_url(self, repository_url: str, version: str) -> str:
        """Return registry url with blob api."""
        return "/".join([repository_url, "blob", self.get_bucket_key(version=version)])

get_blob_url(repository_url, version)

Return registry url with blob api.

Source code in aws_terraform_registry/common/model.py 
45
46
47
def get_blob_url(self, repository_url: str, version: str) -> str:
    """Return registry url with blob api."""
    return "/".join([repository_url, "blob", self.get_bucket_key(version=version)])

get_bucket_key(version)

Return bucket key.

Source code in aws_terraform_registry/common/model.py 
32
33
34
def get_bucket_key(self, version: str) -> str:
    """Return bucket key."""
    return f"{self.module_id}/{version}/{BUCKET_FILE_NAME}"

get_publish_url(bucket_name, version)

Return s3 url.

Source code in aws_terraform_registry/common/model.py 
36
37
38
39
40
41
42
43
def get_publish_url(self, bucket_name: str, version: str) -> str:
    """Return s3 url."""

    region = _find_caller_region()
    bucket_sub_name = f"s3-{region}" if region != "us-east-1" else "s3"
    return "/".join(
        [f"s3::https://{bucket_name}.{bucket_sub_name}.amazonaws.com", self.get_bucket_key(version=version)]
    )

build_parser(config)

Build arguments parser.

Source code in aws_terraform_registry/cli.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def build_parser(config: ApplicationConfig):
    """Build arguments parser."""
    parser = argparse.ArgumentParser(prog="tfr", description="Manage terraform registry")
    subparsers = parser.add_subparsers(help='commands')

    for item in [
        _define_config,
        _define_generate,
        _define_terraformrc,
        _define_release,
        _define_publish,
        _define_unpublish,
    ]:
        item(subparsers, config)

    return parser

generate_terraformrc(config, output_directory, weeks=52)

Generate terraform_rc file.

Parameters:

Name Type Description Default
config ApplicationConfig

application configuration.

 required
output_directory str

directory where to wrote the .terraform_rc.

 required
weeks int

weeks of validity (52 per default).

 52
Source code in aws_terraform_registry/common/token.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def generate_terraformrc(config: ApplicationConfig, output_directory: str, weeks: int = 52):
    """Generate terraform_rc file.

    Args:
        config (ApplicationConfig): application configuration.
        output_directory (str): directory where to wrote the .terraform_rc.
        weeks (int, optional): weeks of validity (52 per default).

    """
    config.validate()
    repository_url = str(config.repository_url)  # avoid mypi error: this could not be null
    hostmane = repository_url[repository_url.index("://") + 3 : len(repository_url)]
    with open(os.path.join(output_directory, ".terraformrc"), "w") as f:
        f.write(
            f"""
credentials "{hostmane}" {{
    token = "{generate_token(config=config, weeks=weeks)}"
}}
"""
        )

generate_token(config, weeks=1)

Generate a JWT tokecm.

Parameters:

Name Type Description Default
config ApplicationConfig

application configuration.

 required
weeks int

weeks of validity (1 per default).

 1

Returns:

Name Type Description
str str

encoded jwt token
Source code in aws_terraform_registry/common/token.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def generate_token(config: ApplicationConfig, weeks: int = 1) -> str:
    """Generate a JWT tokecm.

    Args:
        config (ApplicationConfig): application configuration.
        weeks (int, optional): weeks of validity (1 per default).

    Returns:
        str: encoded jwt token
    """
    config.validate()
    token = get_secret(secret_key_name=config.secret_key_name)  # pyright: ignore[reportArgumentType]
    return jwt.encode(
        {"exp": datetime.datetime.now(tz=datetime.timezone.utc) + datetime.timedelta(weeks=weeks)},
        token,
        algorithm="HS256",
    )

publish_module(config, terraform_module, version, source)

Publish terraform module.

Args:

config (ApplicationConfig): application configuration
terraform_module (TerraformModuleIdentifier): module identifier
version (str): version to publish
source (str): module source

Returns:

Name Type Description
str str

source

Raises:

Type Description
RuntimeError

if specified version and module is ever published.
Source code in aws_terraform_registry/common/publish.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def publish_module(
    config: ApplicationConfig, terraform_module: TerraformModuleIdentifier, version: str, source: str
) -> str:
    """Publish terraform module.

    Args:

        config (ApplicationConfig): application configuration
        terraform_module (TerraformModuleIdentifier): module identifier
        version (str): version to publish
        source (str): module source

    Returns:
        str: source

    Raises:
        (RuntimeError): if specified version and module is ever published.
    """
    config.validate()

    if exists(config=config, terraform_module=terraform_module, version=version):
        msg = f"Version {version} for module {terraform_module.module_id} ever exist"
        logger.error(msg)
        raise RuntimeError(msg)

    client('dynamodb').put_item(
        TableName=config.dynamodb_table_name,
        Item={
            "Id": {"S": terraform_module.module_id},
            "Version": {"S": version},
            "Source": {"S": source},
        },
    )
    logger.info(f"Published module {terraform_module.module_id}, Version {version}, Source {source}")
    return source

release_module(config, terraform_module, version, source)

Release a terraform module.

Source could be:

  • a local folder (In this case local folder will be targzified).
  • an url which point to a targzified archive (like a git release)

This source will be send to the default bucket and publish onto the registry.

Args:

config (ApplicationConfig): application configuration
terraform_module (TerraformModuleIdentifier): module identifier
version (str): version to publish
source (str): module source
Raise

(RuntimeError): if source did not exists or if specified version and module is ever published.

Source code in aws_terraform_registry/common/release.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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
def release_module(
    config: ApplicationConfig, terraform_module: TerraformModuleIdentifier, version: str, source: str
) -> str:
    """Release a terraform module.

    Source could be:

    - a local folder (In this case local folder will be targzified).
    - an url which point to a targzified archive (like a git release)

    This source will be send to the default bucket and publish onto the registry.

    Args:

        config (ApplicationConfig): application configuration
        terraform_module (TerraformModuleIdentifier): module identifier
        version (str): version to publish
        source (str): module source

    Raise:
        (RuntimeError): if source did not exists or if specified version and module is ever published.
    """
    config.validate()
    # remove the v
    version = version if not version.lower().startswith("v") else version[1:]

    if exists(config=config, terraform_module=terraform_module, version=version):
        msg = f"Version {version} for module {terraform_module.module_id} ever exist"
        logger.error(msg)
        raise RuntimeError(msg)

    s3_key = terraform_module.get_bucket_key(version=version)
    logger.debug(f"Put module archive to {s3_key}")

    if source.lower().startswith("http"):
        send_s3_from_url(config=config, source_url=source, s3_key=s3_key)
    else:
        _source = Path(source)
        if not _source.exists():
            raise RuntimeError(f"Source {source} did not exists ")
        if _source.is_file():
            send_s3_from_file(config=config, archive_file=_source, s3_key=s3_key)

        send_s3_from_dir(config=config, archive_dir=_source, s3_key=s3_key)

    publish_url = terraform_module.get_publish_url(
        bucket_name=config.bucket_name, version=version  # pyright: ignore[reportArgumentType]
    )
    # experimental API
    # publish_url = terraform_module.get_blob_url(repository_url=config.repository_url, version=version)
    logger.debug(f"url: {publish_url}")
    publish_module(config=config, terraform_module=terraform_module, version=version, source=publish_url)

    return publish_url

unpublish_module(config, terraform_module, version)

UnPublish terraform module.

Args:

config (ApplicationConfig): application configuration
terraform_module (TerraformModuleIdentifier): module identifier
version (str): version to publish

Raises:

Type Description
RuntimeError

if specified version and module did not exists.
Source code in aws_terraform_registry/common/publish.py 
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
def unpublish_module(config: ApplicationConfig, terraform_module: TerraformModuleIdentifier, version: str):
    """UnPublish terraform module.

    Args:

        config (ApplicationConfig): application configuration
        terraform_module (TerraformModuleIdentifier): module identifier
        version (str): version to publish


    Raises:
        (RuntimeError): if specified version and module did not exists.
    """
    config.validate()

    if not exists(config=config, terraform_module=terraform_module, version=version):
        msg = f"Version {version} for module {terraform_module.module_id} did not exist"
        logger.error(msg)
        raise RuntimeError(msg)

    client('dynamodb').delete_item(
        TableName=config.dynamodb_table_name,
        Key={
            "Id": {"S": terraform_module.module_id},
            "Version": {"S": version},
        },
    )
    logger.info(f"Unpublished module {terraform_module.module_id}, Version {version}")