runway.cfngin.util module

CFNgin utilities.

runway.cfngin.util.camel_to_snake(name: str)str[source]

Convert CamelCase to snake_case.

Parameters

name (str) – The name to convert from CamelCase to snake_case.

Returns

Converted string.

Return type

str

runway.cfngin.util.convert_class_name(kls: type)str[source]

Get a string that represents a given class.

Parameters

kls – The class being analyzed for its name.

Returns

The name of the given kls.

runway.cfngin.util.parse_zone_id(full_zone_id: str)str[source]

Parse the returned hosted zone id and returns only the ID itself.

runway.cfngin.util.get_hosted_zone_by_name(client: Route53Client, zone_name: str) → Optional[str][source]

Get the zone id of an existing zone by name.

Parameters
  • client – The connection used to interact with Route53’s API.

  • zone_name – The name of the DNS hosted zone to create.

Returns

The Id of the Hosted Zone.

runway.cfngin.util.get_or_create_hosted_zone(client: Route53Client, zone_name: str)str[source]

Get the Id of an existing zone, or create it.

Parameters
  • client – The connection used to interact with Route53’s API.

  • zone_name – The name of the DNS hosted zone to create.

Returns

The Id of the Hosted Zone.

class runway.cfngin.util.SOARecordText(record_text: str)[source]

Bases: object

Represents the actual body of an SOARecord.

Instantiate class.

class runway.cfngin.util.SOARecord(record: ResourceRecordSetTypeDef)[source]

Bases: object

Represents an SOA record.

Instantiate class.

runway.cfngin.util.get_soa_record(client: Route53Client, zone_id: str, zone_name: str)SOARecord[source]

Get the SOA record for zone_name from zone_id.

Parameters
  • client – The connection used to interact with Route53’s API.

  • zone_id – The AWS Route53 zone id of the hosted zone to query.

  • zone_name – The name of the DNS hosted zone to create.

Returns

An object representing the parsed SOA record returned from AWS Route53.

runway.cfngin.util.create_route53_zone(client: Route53Client, zone_name: str)str[source]

Create the given zone_name if it doesn’t already exists.

Also sets the SOA negative caching TTL to something short (300 seconds).

Parameters
  • client – The connection used to interact with Route53’s API.

  • zone_name – The name of the DNS hosted zone to create.

Returns

The zone id returned from AWS for the existing, or newly created zone.

runway.cfngin.util.yaml_to_ordered_dict(stream: str, loader: Union[Type[yaml.loader.Loader], Type[yaml.loader.SafeLoader]] = <class 'yaml.loader.SafeLoader'>) → OrderedDict[str, Any][source]

yaml.load alternative with preserved dictionary order.

Parameters
  • stream – YAML string to load.

  • loader – PyYAML loader class. Defaults to safe load.

runway.cfngin.util.uppercase_first_letter(string_: str)str[source]

Return string with first character upper case.

runway.cfngin.util.cf_safe_name(name: str)str[source]

Convert a name to a safe string for a CloudFormation resource.

Given a string, returns a name that is safe for use as a CloudFormation Resource. (ie: Only alphanumeric characters)

runway.cfngin.util.read_value_from_path(value: str, *, root_path: Optional[pathlib.Path] = None)str[source]

Enable translators to read values from files.

The value can be referred to with the file:// prefix.

Example

conf_key: ${kms file://kms_value.txt}
runway.cfngin.util.get_client_region(client: Any)str[source]

Get the region from a boto3 client.

Parameters

client – The client to get the region from.

Returns

AWS region string.

runway.cfngin.util.get_s3_endpoint(client: Any)str[source]

Get the s3 endpoint for the given boto3 client.

Parameters

client – The client to get the endpoint from.

Returns

The AWS endpoint for the client.

runway.cfngin.util.s3_bucket_location_constraint(region: Optional[str]) → Optional[str][source]

Return the appropriate LocationConstraint info for a new S3 bucket.

When creating a bucket in a region OTHER than us-east-1, you need to specify a LocationConstraint inside the CreateBucketConfiguration argument. This function helps you determine the right value given a given client.

Parameters

region – The region where the bucket will be created in.

Returns

The string to use with the given client for creating a bucket.

runway.cfngin.util.ensure_s3_bucket(s3_client: S3Client, bucket_name: str, bucket_region: Optional[str] = None, persist_graph: bool = False)None[source]

Ensure an s3 bucket exists, if it does not then create it.

Parameters
  • s3_client – An s3 client used to verify and create the bucket.

  • bucket_name – The bucket being checked/created.

  • bucket_region – The region to create the bucket in. If not provided, will be determined by s3_client’s region.

  • persist_graph – Check bucket for recommended settings. If creating a new bucket, it will be created with recommended settings.

runway.cfngin.util.parse_cloudformation_template(template: str) → Dict[str, Any][source]

Parse CFN template string.

Leverages the vendored aws-cli yamlhelper to handle JSON or YAML templates.

Parameters

template – The template body.

class runway.cfngin.util.Extractor(archive: Optional[pathlib.Path] = None)[source]

Bases: object

Base class for extractors.

Instantiate class.

Parameters

archive (str) – Archive path.

extension: ClassVar[str] = ''
set_archive(dir_name: pathlib.Path)None[source]

Update archive filename to match directory name & extension.

Parameters

dir_name – Archive directory name

class runway.cfngin.util.TarExtractor(archive: Optional[pathlib.Path] = None)[source]

Bases: runway.cfngin.util.Extractor

Extracts tar archives.

Instantiate class.

Parameters

archive (str) – Archive path.

extension: ClassVar[str] = '.tar'
extract(destination: pathlib.Path)None[source]

Extract the archive.

class runway.cfngin.util.TarGzipExtractor(archive: Optional[pathlib.Path] = None)[source]

Bases: runway.cfngin.util.Extractor

Extracts compressed tar archives.

Instantiate class.

Parameters

archive (str) – Archive path.

extension: ClassVar[str] = '.tar.gz'
extract(destination: pathlib.Path)None[source]

Extract the archive.

class runway.cfngin.util.ZipExtractor(archive: Optional[pathlib.Path] = None)[source]

Bases: runway.cfngin.util.Extractor

Extracts zip archives.

Instantiate class.

Parameters

archive (str) – Archive path.

extension: ClassVar[str] = '.zip'
extract(destination: pathlib.Path)None[source]

Extract the archive.

class runway.cfngin.util.SourceProcessor(sources: CfnginPackageSourcesDefinitionModel, cache_dir: Optional[Path] = None)[source]

Bases: object

Makes remote python package sources available in current environment.

Process a config’s defined package sources.

Parameters
  • sources – Package sources from CFNgin config dictionary.

  • cache_dir – Path where remote sources will be cached.

ISO8601_FORMAT = '%Y%m%dT%H%M%SZ'
create_cache_directories()None[source]

Ensure that SourceProcessor cache directories exist.

get_package_sources()None[source]

Make remote python packages available for local use.

fetch_local_package(config: LocalCfnginPackageSourceDefinitionModel)None[source]

Make a local path available to current CFNgin config.

Parameters

config – Package source config.

fetch_s3_package(config: S3CfnginPackageSourceDefinitionModel)None[source]

Make a remote S3 archive available for local use.

Parameters

config – Package source config.

fetch_git_package(config: GitCfnginPackageSourceDefinitionModel)None[source]

Make a remote git repository available for local use.

Parameters

config – Package source config.

update_paths_and_config(config: Union[GitCfnginPackageSourceDefinitionModel, LocalCfnginPackageSourceDefinitionModel, S3CfnginPackageSourceDefinitionModel], pkg_dir_name: str, pkg_cache_dir: Optional[Path] = None)None[source]

Handle remote source defined sys.paths & configs.

Parameters
  • config – Package source config.

  • pkg_dir_name – Directory name of the CFNgin archive.

  • pkg_cache_dir – Fully qualified path to CFNgin cache cache directory.

static git_ls_remote(uri: str, ref: str)str[source]

Determine the latest commit id for a given ref.

Parameters
  • uri – Git URI.

  • ref – Git ref.

static determine_git_ls_remote_ref(config: GitCfnginPackageSourceDefinitionModel)str[source]

Determine the ref to be used with the “git ls-remote” command.

Parameters

config – Git package source config.

Returns

A branch reference or “HEAD”.

determine_git_ref(config: GitCfnginPackageSourceDefinitionModel)str[source]

Determine the ref to be used for git checkout.

Parameters

config – Git package source config.

Returns

A commit id or tag name.

static sanitize_uri_path(uri: str)str[source]

Take a URI and converts it to a directory safe path.

Parameters

uri – URI (e.g. http://example.com/cats).

Returns

Directory name for the supplied uri.

sanitize_git_path(uri: str, ref: Optional[str] = None)str[source]

Take a git URI and ref and converts it to a directory safe path.

Parameters
  • uri – Git URI. (e.g. git@github.com:foo/bar.git)

  • ref – Git ref to be appended to the path.

Returns

Directory name for the supplied uri

runway.cfngin.util.stack_template_key_name(blueprint: Blueprint)str[source]

Given a blueprint, produce an appropriate key name.

Parameters

blueprint – The blueprint object to create the key from.

Returns

Key name resulting from blueprint.