Mercurial > repos > shellac > guppy_basecaller
diff env/lib/python3.7/site-packages/boto/cloudformation/stack.py @ 0:26e78fe6e8c4 draft
"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
author | shellac |
---|---|
date | Sat, 02 May 2020 07:14:21 -0400 |
parents | |
children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/env/lib/python3.7/site-packages/boto/cloudformation/stack.py Sat May 02 07:14:21 2020 -0400 @@ -0,0 +1,423 @@ +from datetime import datetime + +from boto.resultset import ResultSet + + +class Stack(object): + def __init__(self, connection=None): + self.connection = connection + self.creation_time = None + self.description = None + self.disable_rollback = None + self.notification_arns = [] + self.outputs = [] + self.parameters = [] + self.capabilities = [] + self.tags = [] + self.stack_id = None + self.stack_status = None + self.stack_status_reason = None + self.stack_name = None + self.timeout_in_minutes = None + + @property + def stack_name_reason(self): + return self.stack_status_reason + + @stack_name_reason.setter + def stack_name_reason(self, value): + self.stack_status_reason = value + + def startElement(self, name, attrs, connection): + if name == "Parameters": + self.parameters = ResultSet([('member', Parameter)]) + return self.parameters + elif name == "Outputs": + self.outputs = ResultSet([('member', Output)]) + return self.outputs + elif name == "Capabilities": + self.capabilities = ResultSet([('member', Capability)]) + return self.capabilities + elif name == "Tags": + self.tags = Tag() + return self.tags + elif name == 'NotificationARNs': + self.notification_arns = ResultSet([('member', NotificationARN)]) + return self.notification_arns + else: + return None + + def endElement(self, name, value, connection): + if name == 'CreationTime': + try: + self.creation_time = datetime.strptime(value, '%Y-%m-%dT%H:%M:%SZ') + except ValueError: + self.creation_time = datetime.strptime(value, '%Y-%m-%dT%H:%M:%S.%fZ') + elif name == "Description": + self.description = value + elif name == "DisableRollback": + if str(value).lower() == 'true': + self.disable_rollback = True + else: + self.disable_rollback = False + elif name == 'StackId': + self.stack_id = value + elif name == 'StackName': + self.stack_name = value + elif name == 'StackStatus': + self.stack_status = value + elif name == "StackStatusReason": + self.stack_status_reason = value + elif name == "TimeoutInMinutes": + self.timeout_in_minutes = int(value) + elif name == "member": + pass + else: + setattr(self, name, value) + + def delete(self): + return self.connection.delete_stack(stack_name_or_id=self.stack_id) + + def describe_events(self, next_token=None): + return self.connection.describe_stack_events( + stack_name_or_id=self.stack_id, + next_token=next_token + ) + + def describe_resource(self, logical_resource_id): + return self.connection.describe_stack_resource( + stack_name_or_id=self.stack_id, + logical_resource_id=logical_resource_id + ) + + def describe_resources(self, logical_resource_id=None, + physical_resource_id=None): + return self.connection.describe_stack_resources( + stack_name_or_id=self.stack_id, + logical_resource_id=logical_resource_id, + physical_resource_id=physical_resource_id + ) + + def list_resources(self, next_token=None): + return self.connection.list_stack_resources( + stack_name_or_id=self.stack_id, + next_token=next_token + ) + + def update(self): + rs = self.connection.describe_stacks(self.stack_id) + if len(rs) == 1 and rs[0].stack_id == self.stack_id: + self.__dict__.update(rs[0].__dict__) + else: + raise ValueError("%s is not a valid Stack ID or Name" % + self.stack_id) + + def get_template(self): + return self.connection.get_template(stack_name_or_id=self.stack_id) + + def get_policy(self): + """ + Returns the stack policy for this stack. If it has no policy + then, a null value is returned. + """ + return self.connection.get_stack_policy(self.stack_id) + + def set_policy(self, stack_policy_body=None, stack_policy_url=None): + """ + Sets a stack policy for this stack. + + :type stack_policy_body: string + :param stack_policy_body: Structure containing the stack policy body. + (For more information, go to ` Prevent Updates to Stack Resources`_ + in the AWS CloudFormation User Guide.) + You must pass `StackPolicyBody` or `StackPolicyURL`. If both are + passed, only `StackPolicyBody` is used. + + :type stack_policy_url: string + :param stack_policy_url: Location of a file containing the stack + policy. The URL must point to a policy (max size: 16KB) located in + an S3 bucket in the same region as the stack. You must pass + `StackPolicyBody` or `StackPolicyURL`. If both are passed, only + `StackPolicyBody` is used. + """ + return self.connection.set_stack_policy(self.stack_id, + stack_policy_body=stack_policy_body, + stack_policy_url=stack_policy_url) + + +class StackSummary(object): + def __init__(self, connection=None): + self.connection = connection + self.stack_id = None + self.stack_status = None + self.stack_name = None + self.creation_time = None + self.deletion_time = None + self.template_description = None + + def startElement(self, name, attrs, connection): + return None + + def endElement(self, name, value, connection): + if name == 'StackId': + self.stack_id = value + elif name == 'StackStatus': + self.stack_status = value + elif name == 'StackName': + self.stack_name = value + elif name == 'CreationTime': + try: + self.creation_time = datetime.strptime(value, '%Y-%m-%dT%H:%M:%SZ') + except ValueError: + self.creation_time = datetime.strptime(value, '%Y-%m-%dT%H:%M:%S.%fZ') + elif name == "DeletionTime": + try: + self.deletion_time = datetime.strptime(value, '%Y-%m-%dT%H:%M:%SZ') + except ValueError: + self.deletion_time = datetime.strptime(value, '%Y-%m-%dT%H:%M:%S.%fZ') + elif name == 'TemplateDescription': + self.template_description = value + elif name == "member": + pass + else: + setattr(self, name, value) + + +class Parameter(object): + def __init__(self, connection=None): + self.connection = None + self.key = None + self.value = None + + def startElement(self, name, attrs, connection): + return None + + def endElement(self, name, value, connection): + if name == "ParameterKey": + self.key = value + elif name == "ParameterValue": + self.value = value + else: + setattr(self, name, value) + + def __repr__(self): + return "Parameter:\"%s\"=\"%s\"" % (self.key, self.value) + + +class Output(object): + def __init__(self, connection=None): + self.connection = connection + self.description = None + self.key = None + self.value = None + + def startElement(self, name, attrs, connection): + return None + + def endElement(self, name, value, connection): + if name == "Description": + self.description = value + elif name == "OutputKey": + self.key = value + elif name == "OutputValue": + self.value = value + else: + setattr(self, name, value) + + def __repr__(self): + return "Output:\"%s\"=\"%s\"" % (self.key, self.value) + + +class Capability(object): + def __init__(self, connection=None): + self.connection = None + self.value = None + + def startElement(self, name, attrs, connection): + return None + + def endElement(self, name, value, connection): + self.value = value + + def __repr__(self): + return "Capability:\"%s\"" % (self.value) + + +class Tag(dict): + + def __init__(self, connection=None): + dict.__init__(self) + self.connection = connection + self._current_key = None + self._current_value = None + + def startElement(self, name, attrs, connection): + return None + + def endElement(self, name, value, connection): + if name == "Key": + self._current_key = value + elif name == "Value": + self._current_value = value + else: + setattr(self, name, value) + + if self._current_key and self._current_value: + self[self._current_key] = self._current_value + self._current_key = None + self._current_value = None + + +class NotificationARN(object): + def __init__(self, connection=None): + self.connection = None + self.value = None + + def startElement(self, name, attrs, connection): + return None + + def endElement(self, name, value, connection): + self.value = value + + def __repr__(self): + return "NotificationARN:\"%s\"" % (self.value) + + +class StackResource(object): + def __init__(self, connection=None): + self.connection = connection + self.description = None + self.logical_resource_id = None + self.physical_resource_id = None + self.resource_status = None + self.resource_status_reason = None + self.resource_type = None + self.stack_id = None + self.stack_name = None + self.timestamp = None + + def startElement(self, name, attrs, connection): + return None + + def endElement(self, name, value, connection): + if name == "Description": + self.description = value + elif name == "LogicalResourceId": + self.logical_resource_id = value + elif name == "PhysicalResourceId": + self.physical_resource_id = value + elif name == "ResourceStatus": + self.resource_status = value + elif name == "ResourceStatusReason": + self.resource_status_reason = value + elif name == "ResourceType": + self.resource_type = value + elif name == "StackId": + self.stack_id = value + elif name == "StackName": + self.stack_name = value + elif name == "Timestamp": + try: + self.timestamp = datetime.strptime(value, '%Y-%m-%dT%H:%M:%SZ') + except ValueError: + self.timestamp = datetime.strptime(value, '%Y-%m-%dT%H:%M:%S.%fZ') + else: + setattr(self, name, value) + + def __repr__(self): + return "StackResource:%s (%s)" % (self.logical_resource_id, + self.resource_type) + + +class StackResourceSummary(object): + def __init__(self, connection=None): + self.connection = connection + self.last_updated_time = None + self.logical_resource_id = None + self.physical_resource_id = None + self.resource_status = None + self.resource_status_reason = None + self.resource_type = None + + def startElement(self, name, attrs, connection): + return None + + def endElement(self, name, value, connection): + if name == "LastUpdatedTime": + try: + self.last_updated_time = datetime.strptime( + value, + '%Y-%m-%dT%H:%M:%SZ' + ) + except ValueError: + self.last_updated_time = datetime.strptime( + value, + '%Y-%m-%dT%H:%M:%S.%fZ' + ) + elif name == "LogicalResourceId": + self.logical_resource_id = value + elif name == "PhysicalResourceId": + self.physical_resource_id = value + elif name == "ResourceStatus": + self.resource_status = value + elif name == "ResourceStatusReason": + self.resource_status_reason = value + elif name == "ResourceType": + self.resource_type = value + else: + setattr(self, name, value) + + def __repr__(self): + return "StackResourceSummary:%s (%s)" % (self.logical_resource_id, + self.resource_type) + + +class StackEvent(object): + valid_states = ("CREATE_IN_PROGRESS", "CREATE_FAILED", "CREATE_COMPLETE", + "DELETE_IN_PROGRESS", "DELETE_FAILED", "DELETE_COMPLETE") + def __init__(self, connection=None): + self.connection = connection + self.event_id = None + self.logical_resource_id = None + self.physical_resource_id = None + self.resource_properties = None + self.resource_status = None + self.resource_status_reason = None + self.resource_type = None + self.stack_id = None + self.stack_name = None + self.timestamp = None + + def startElement(self, name, attrs, connection): + return None + + def endElement(self, name, value, connection): + if name == "EventId": + self.event_id = value + elif name == "LogicalResourceId": + self.logical_resource_id = value + elif name == "PhysicalResourceId": + self.physical_resource_id = value + elif name == "ResourceProperties": + self.resource_properties = value + elif name == "ResourceStatus": + self.resource_status = value + elif name == "ResourceStatusReason": + self.resource_status_reason = value + elif name == "ResourceType": + self.resource_type = value + elif name == "StackId": + self.stack_id = value + elif name == "StackName": + self.stack_name = value + elif name == "Timestamp": + try: + self.timestamp = datetime.strptime(value, '%Y-%m-%dT%H:%M:%SZ') + except ValueError: + self.timestamp = datetime.strptime(value, '%Y-%m-%dT%H:%M:%S.%fZ') + else: + setattr(self, name, value) + + def __repr__(self): + return "StackEvent %s %s %s" % (self.resource_type, + self.logical_resource_id, self.resource_status)