PK!DjQ@@troposphere_ez/__init__.pyfrom troposphere_ez._stack_manager_base import StackManagerBase PK! 1M M %troposphere_ez/_stack_manager_base.pyimport json import os import os.path import shlex import subprocess from collections import OrderedDict from tempfile import TemporaryDirectory from termcolor import cprint from typing import List, Optional from troposphere import Output, Template class StackManagerBase: aws_profile: str stage: str prefix: str _template: Template _outputs: List[Output] _output_results: Optional[OrderedDict] def __init__(self, aws_profile: str, stage: str, prefix: str): self.aws_profile = aws_profile self.stage = stage self.prefix = prefix self._template = Template() self._outputs = [] self._output_results = None def rec_to_output(self, title, value): self._outputs.append(Output(title, Value=value)) def add_resource(self, res): self._template.add_resource(res) def toyaml(self): self._template.add_output(self._outputs) return self._template.to_yaml() def apply(self): self._template.add_output(self._outputs) with TemporaryDirectory() as tmpdir: the_file = os.path.join(tmpdir, "template.yml") with open(the_file, "wt") as fp: fp.write(self._template.to_yaml()) cmd_env = os.environ.copy() # AWS_SDK_LOAD_CONFIG をセットしていないと~/.aws/configのregsionを読んでくれない cmd_env["AWS_SDK_LOAD_CONFIG"] = "true" cmd_args = [ "cfn-create-or-update", "--profile", self.aws_profile, "--template-body", f"file://{the_file}", "--capabilities", "CAPABILITY_NAMED_IAM", "--stack-name", f"{self.prefix}-{self.stage}", "--wait", ] print("$ ", end="") cprint(" ".join([shlex.quote(s) for s in cmd_args]), "cyan") subprocess.check_output(cmd_args, env=cmd_env) self._sync_output_results() def _run_aws_cli(self, *rest_args) -> str: head_args = ["aws", "--profile", self.aws_profile] full_args = head_args + list(rest_args) print("$ ", end="") cprint(" ".join([shlex.quote(s) for s in full_args]), "cyan") ret = subprocess.check_output(full_args) return ret.decode().strip() def _sync_output_results(self): """CloudFormationのOutputの結果を辞書形式で得てインスタンス変数に格納。 """ ret = self._run_aws_cli( "cloudformation", "describe-stacks", "--stack-name", f"{self.prefix}-{self.stage}", ) rows = json.loads(ret)["Stacks"][0]["Outputs"] output_results_lst = [(row["OutputKey"], row["OutputValue"]) for row in rows] output_results_lst.sort(key=lambda x: x[0]) self._output_results = OrderedDict(output_results_lst) @property def output_results(self) -> OrderedDict: return self._output_results def pprint_output_results(self): """output_resultsを色付け&フォーマットして表示する """ if self._output_results is None: return for k, v in self._output_results.items(): cprint(f"{k}", "green", end="") print(f": {v}") return self._output_results PK!4Rtroposphere_ez/kit/__init__.pyfrom troposphere_ez.kit._dynamodb_simple_kit import DynamodbSimpleKit from troposphere_ez.kit._s3_simple_kit import S3SimpleKit from troposphere_ez.kit._sqs_simple_kit import SqsSimpleKit PK!>F*troposphere_ez/kit/_dynamodb_simple_kit.pyfrom troposphere import dynamodb, Ref, GetAtt from troposphere_ez.utils import pascalyze_atom from typing import Dict class DynamodbSimpleKit: def add_simple_dynamodb_table( self, atom: str, attr_definitions: Dict[str, str], key_schema: Dict[str, str], rcu: int, wcu: int, output=True, ) -> dynamodb.Table: patom = pascalyze_atom(atom) dtbl = dynamodb.Table(title=f"SimpleDynamodbTable{patom}") dtbl.AttributeDefinitions = [ dynamodb.AttributeDefinition(AttributeName=name, AttributeType=atype) for (name, atype) in attr_definitions.items() ] dtbl.KeySchema = [ dynamodb.KeySchema(AttributeName=name, KeyType=ktype) for (name, ktype) in key_schema.items() ] dtbl.ProvisionedThroughput = dynamodb.ProvisionedThroughput( ReadCapacityUnits=rcu, WriteCapacityUnits=wcu ) self.add_resource(dtbl) if output: self.rec_to_output(dtbl.title, Ref(dtbl)) self.rec_to_output(dtbl.title + "Arn", GetAtt(dtbl, "Arn")) return dtbl PK! 3$troposphere_ez/kit/_s3_simple_kit.pyfrom troposphere import s3, Output, Ref, GetAtt from troposphere_ez.utils import pascalyze_atom class S3SimpleKit: def add_secret_s3_bucket(self, atom: str, output=True) -> s3.Bucket: "非公開なs3バケット" patom = pascalyze_atom(atom) bucket = s3.Bucket(f"SecretS3Bucket{patom}") bucket.AccessControl = "Private" bucket.PublicAccessBlockConfiguration = s3.PublicAccessBlockConfiguration( BlockPublicAcls=True, BlockPublicPolicy=True, IgnorePublicAcls=True, RestrictPublicBuckets=True, ) self.add_resource(bucket) if output: self.rec_to_output(bucket.title, Ref(bucket)) self.rec_to_output(bucket.title + "Arn", GetAtt(bucket, "Arn")) return bucket def add_public_s3_bucket(self, atom: str, output=True) -> s3.Bucket: "公開なs3バケット" patom = pascalyze_atom(atom) bucket = s3.Bucket(f"PublicS3Bucket{patom}") bucket.AccessControl = "PublicRead" bucket.PublicAccessBlockConfiguration = s3.PublicAccessBlockConfiguration( BlockPublicAcls=False, BlockPublicPolicy=False, IgnorePublicAcls=False, RestrictPublicBuckets=False, ) self.add_resource(bucket) if output: self.rec_to_output(bucket.title, Ref(bucket)) self.rec_to_output(bucket.title + "Arn", GetAtt(bucket, "Arn")) return bucket PK!7%//%troposphere_ez/kit/_sqs_simple_kit.pyfrom troposphere import sqs, Ref, GetAtt from troposphere_ez.utils import pascalyze_atom class SqsSimpleKit: def add_fifo_sqs_queue(self, atom: str, output=True) -> sqs.Queue: patom = pascalyze_atom(atom) que = sqs.Queue(title=f"FifoSqsQueue{patom}") que.FifoQueue = True que.ContentBasedDeduplication = True self.add_resource(que) if output: self.rec_to_output(que.title, Ref(que)) self.rec_to_output(que.title + "Arn", GetAtt(que, "Arn")) self.rec_to_output(que.title + "QueueName", GetAtt(que, "QueueName")) return que def add_simple_standard_sqs_queue(self, atom: str, output=True) -> sqs.Queue: patom = pascalyze_atom(atom) que = sqs.Queue(title=f"StandardSqsQueue{patom}") self.add_resource(que) if output: self.rec_to_output(que.title, Ref(que)) self.rec_to_output(que.title + "Arn", GetAtt(que, "Arn")) self.rec_to_output(que.title + "QueueName", GetAtt(que, "QueueName")) return que PK!troposphere_ez/marker.pyPK!*tttroposphere_ez/utils.pyimport re from stringcase import pascalcase from troposphere import Join def concat(*args): return Join("", args) def pascalyze_atom(atom: str) -> str: """hoge-fooのような名前をHogeFooへ置き換える """ if not re.search(r"^[a-z][a-z-]*$", atom) or atom[-1] == "-": raise ValueError() return pascalcase(atom.replace("-", "_")) PK!HnHTU$troposphere_ez-0.1.0.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!HמD'troposphere_ez-0.1.0.dist-info/METADATAQO0+&[dž"ihDDAKWufz;4J1;wVxF*troposphere_ez/kit/_dynamodb_simple_kit.pyPK! 3$troposphere_ez/kit/_s3_simple_kit.pyPK!7%//%troposphere_ez/kit/_sqs_simple_kit.pyPK!Ytroposphere_ez/marker.pyPK!*tttroposphere_ez/utils.pyPK!HnHTU$8 troposphere_ez-0.1.0.dist-info/WHEELPK!HמD' troposphere_ez-0.1.0.dist-info/METADATAPK!HMD%W"troposphere_ez-0.1.0.dist-info/RECORDPK i$