<s> from __future__ import unicode_literals <EOL> import copy <EOL> import boto . rds <EOL> from jinja2 import Template <EOL> from moto . cloudformation . exceptions import UnformattedGetAttTemplateException <EOL> from moto . core import BaseBackend <EOL> from moto . core . utils import get_random_hex <EOL> from moto . ec2 . models import ec2_backends <EOL> from . exceptions import DBInstanceNotFoundError , DBSecurityGroupNotFoundError , DBSubnetGroupNotFoundError <EOL> class Database ( object ) : <EOL> def __init__ ( self , ** kwargs ) : <EOL> self . status = "<STR_LIT>" <EOL> self . is_replica = False <EOL> self . replicas = [ ] <EOL> self . region = kwargs . get ( '<STR_LIT>' ) <EOL> self . engine = kwargs . get ( "<STR_LIT>" ) <EOL> self . engine_version = kwargs . get ( "<STR_LIT>" ) <EOL> if self . engine_version is None : <EOL> self . engine_version = "<STR_LIT>" <EOL> self . iops = kwargs . get ( "<STR_LIT>" ) <EOL> self . storage_type = kwargs . get ( "<STR_LIT>" ) <EOL> self . master_username = kwargs . get ( '<STR_LIT>' ) <EOL> self . master_password = kwargs . get ( '<STR_LIT>' ) <EOL> self . auto_minor_version_upgrade = kwargs . get ( '<STR_LIT>' ) <EOL> if self . auto_minor_version_upgrade is None : <EOL> self . auto_minor_version_upgrade = True <EOL> self . allocated_storage = kwargs . get ( '<STR_LIT>' ) <EOL> self . db_instance_identifier = kwargs . get ( '<STR_LIT>' ) <EOL> self . source_db_identifier = kwargs . get ( "<STR_LIT>" ) <EOL> self . db_instance_class = kwargs . get ( '<STR_LIT>' ) <EOL> self . port = kwargs . get ( '<STR_LIT:port>' ) <EOL> self . db_instance_identifier = kwargs . get ( '<STR_LIT>' ) <EOL> self . db_name = kwargs . get ( "<STR_LIT>" ) <EOL> self . publicly_accessible = kwargs . get ( "<STR_LIT>" ) <EOL> if self . publicly_accessible is None : <EOL> self . publicly_accessible = True <EOL> self . backup_retention_period = kwargs . get ( "<STR_LIT>" ) <EOL> if self . backup_retention_period is None : <EOL> self . backup_retention_period = <NUM_LIT:1> <EOL> self . availability_zone = kwargs . get ( "<STR_LIT>" ) <EOL> self . multi_az = kwargs . get ( "<STR_LIT>" ) <EOL> self . db_subnet_group_name = kwargs . get ( "<STR_LIT>" ) <EOL> if self . db_subnet_group_name : <EOL> self . db_subnet_group = rds_backends [ self . region ] . describe_subnet_groups ( self . db_subnet_group_name ) [ <NUM_LIT:0> ] <EOL> else : <EOL> self . db_subnet_group = [ ] <EOL> self . security_groups = kwargs . get ( '<STR_LIT>' , [ ] ) <EOL> @ property <EOL> def address ( self ) : <EOL> return "<STR_LIT>" . format ( self . db_instance_identifier , self . region ) <EOL> def add_replica ( self , replica ) : <EOL> self . replicas . append ( replica . db_instance_identifier ) <EOL> def remove_replica ( self , replica ) : <EOL> self . replicas . remove ( replica . db_instance_identifier ) <EOL> def set_as_replica ( self ) : <EOL> self . is_replica = True <EOL> self . replicas = [ ] <EOL> def update ( self , db_kwargs ) : <EOL> for key , value in db_kwargs . items ( ) : <EOL> if value is not None : <EOL> setattr ( self , key , value ) <EOL> def get_cfn_attribute ( self , attribute_name ) : <EOL> if attribute_name == '<STR_LIT>' : <EOL> return self . address <EOL> elif attribute_name == '<STR_LIT>' : <EOL> return self . port <EOL> raise UnformattedGetAttTemplateException ( ) <EOL> @ classmethod <EOL> def create_from_cloudformation_json ( cls , resource_name , cloudformation_json , region_name ) : <EOL> properties = cloudformation_json [ '<STR_LIT>' ] <EOL> db_instance_identifier = properties . get ( '<STR_LIT>' ) <EOL> if not db_instance_identifier : <EOL> db_instance_identifier = resource_name . lower ( ) + get_random_hex ( <NUM_LIT:12> ) <EOL> db_security_groups = properties . get ( '<STR_LIT>' ) <EOL> if not db_security_groups : <EOL> db_security_groups = [ ] <EOL> security_groups = [ group . group_name for group in db_security_groups ] <EOL> db_subnet_group = properties . get ( "<STR_LIT>" ) <EOL> db_subnet_group_name = db_subnet_group . subnet_name if db_subnet_group else None <EOL> db_kwargs = { <EOL> "<STR_LIT>" : properties . get ( '<STR_LIT>' ) , <EOL> "<STR_LIT>" : properties . get ( '<STR_LIT>' ) , <EOL> "<STR_LIT>" : properties . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" : properties . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" : properties . get ( '<STR_LIT>' ) , <EOL> "<STR_LIT>" : db_instance_identifier , <EOL> "<STR_LIT>" : properties . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" : db_subnet_group_name , <EOL> "<STR_LIT>" : properties . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" : properties . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" : properties . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" : properties . get ( '<STR_LIT>' ) , <EOL> "<STR_LIT>" : properties . get ( '<STR_LIT>' ) , <EOL> "<STR_LIT>" : properties . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT:port>" : properties . get ( '<STR_LIT>' , <NUM_LIT> ) , <EOL> "<STR_LIT>" : properties . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" : region_name , <EOL> "<STR_LIT>" : security_groups , <EOL> "<STR_LIT>" : properties . get ( "<STR_LIT>" ) , <EOL> } <EOL> rds_backend = rds_backends [ region_name ] <EOL> source_db_identifier = properties . get ( "<STR_LIT>" ) <EOL> if source_db_identifier : <EOL> db_kwargs [ "<STR_LIT>" ] = source_db_identifier . db_instance_identifier <EOL> database = rds_backend . create_database_replica ( db_kwargs ) <EOL> else : <EOL> database = rds_backend . create_database ( db_kwargs ) <EOL> return database <EOL> def to_xml ( self ) : <EOL> template = Template ( """<STR_LIT>""" ) <EOL> return template . render ( database = self ) <EOL> class SecurityGroup ( object ) : <EOL> def __init__ ( self , group_name , description ) : <EOL> self . group_name = group_name <EOL> self . description = description <EOL> self . status = "<STR_LIT>" <EOL> self . ip_ranges = [ ] <EOL> self . ec2_security_groups = [ ] <EOL> def to_xml ( self ) : <EOL> template = Template ( """<STR_LIT>""" ) <EOL> return template . render ( security_group = self ) <EOL> def authorize_cidr ( self , cidr_ip ) : <EOL> self . ip_ranges . append ( cidr_ip ) <EOL> def authorize_security_group ( self , security_group ) : <EOL> self . ec2_security_groups . append ( security_group ) <EOL> @ classmethod <EOL> def create_from_cloudformation_json ( cls , resource_name , cloudformation_json , region_name ) : <EOL> properties = cloudformation_json [ '<STR_LIT>' ] <EOL> group_name = resource_name . lower ( ) + get_random_hex ( <NUM_LIT:12> ) <EOL> description = properties [ '<STR_LIT>' ] <EOL> security_group_ingress = properties [ '<STR_LIT>' ] <EOL> ec2_backend = ec2_backends [ region_name ] <EOL> rds_backend = rds_backends [ region_name ] <EOL> security_group = rds_backend . create_security_group ( <EOL> group_name , <EOL> description , <EOL> ) <EOL> for ingress_type , ingress_value in security_group_ingress . items ( ) : <EOL> if ingress_type == "<STR_LIT>" : <EOL> security_group . authorize_cidr ( ingress_value ) <EOL> elif ingress_type == "<STR_LIT>" : <EOL> subnet = ec2_backend . get_security_group_from_name ( ingress_value ) <EOL> security_group . authorize_security_group ( subnet ) <EOL> elif ingress_type == "<STR_LIT>" : <EOL> subnet = ec2_backend . get_security_group_from_id ( ingress_value ) <EOL> security_group . authorize_security_group ( subnet ) <EOL> return security_group <EOL> class SubnetGroup ( object ) : <EOL> def __init__ ( self , subnet_name , description , subnets ) : <EOL> self . subnet_name = subnet_name <EOL> self . description = description <EOL> self . subnets = subnets <EOL> self . status = "<STR_LIT>" <EOL> self . vpc_id = self . subnets [ <NUM_LIT:0> ] . vpc_id <EOL> def to_xml ( self ) : <EOL> template = Template ( """<STR_LIT>""" ) <EOL> return template . render ( subnet_group = self ) <EOL> @ classmethod <EOL> def create_from_cloudformation_json ( cls , resource_name , cloudformation_json , region_name ) : <EOL> properties = cloudformation_json [ '<STR_LIT>' ] <EOL> subnet_name = resource_name . lower ( ) + get_random_hex ( <NUM_LIT:12> ) <EOL> description = properties [ '<STR_LIT>' ] <EOL> subnet_ids = properties [ '<STR_LIT>' ] <EOL> ec2_backend = ec2_backends [ region_name ] <EOL> subnets = [ ec2_backend . get_subnet ( subnet_id ) for subnet_id in subnet_ids ] <EOL> rds_backend = rds_backends [ region_name ] <EOL> subnet_group = rds_backend . create_subnet_group ( <EOL> subnet_name , <EOL> description , <EOL> subnets , <EOL> ) <EOL> return subnet_group <EOL> class RDSBackend ( BaseBackend ) : <EOL> def __init__ ( self ) : <EOL> self . databases = { } <EOL> self . security_groups = { } <EOL> self . subnet_groups = { } <EOL> def create_database ( self , db_kwargs ) : <EOL> database_id = db_kwargs [ '<STR_LIT>' ] <EOL> database = Database ( ** db_kwargs ) <EOL> self . databases [ database_id ] = database <EOL> return database <EOL> def create_database_replica ( self , db_kwargs ) : <EOL> database_id = db_kwargs [ '<STR_LIT>' ] <EOL> source_database_id = db_kwargs [ '<STR_LIT>' ] <EOL> primary = self . describe_databases ( source_database_id ) [ <NUM_LIT:0> ] <EOL> replica = copy . deepcopy ( primary ) <EOL> replica . update ( db_kwargs ) <EOL> replica . set_as_replica ( ) <EOL> self . databases [ database_id ] = replica <EOL> primary . add_replica ( replica ) <EOL> return replica <EOL> def describe_databases ( self , db_instance_identifier = None ) : <EOL> if db_instance_identifier : <EOL> if db_instance_identifier in self . databases : <EOL> return [ self . databases [ db_instance_identifier ] ] <EOL> else : <EOL> raise DBInstanceNotFoundError ( db_instance_identifier ) <EOL> return self . databases . values ( ) <EOL> def modify_database ( self , db_instance_identifier , db_kwargs ) : <EOL> database = self . describe_databases ( db_instance_identifier ) [ <NUM_LIT:0> ] <EOL> database . update ( db_kwargs ) <EOL> return database <EOL> def delete_database ( self , db_instance_identifier ) : <EOL> if db_instance_identifier in self . databases : <EOL> database = self . databases . pop ( db_instance_identifier ) <EOL> if database . is_replica : <EOL> primary = self . describe_databases ( database . source_db_identifier ) [ <NUM_LIT:0> ] <EOL> primary . remove_replica ( database ) <EOL> return database <EOL> else : <EOL> raise DBInstanceNotFoundError ( db_instance_identifier ) <EOL> def create_security_group ( self , group_name , description ) : <EOL> security_group = SecurityGroup ( group_name , description ) <EOL> self . security_groups [ group_name ] = security_group <EOL> return security_group <EOL> def describe_security_groups ( self , security_group_name ) : <EOL> if security_group_name : <EOL> if security_group_name in self . security_groups : <EOL> return [ self . security_groups [ security_group_name ] ] <EOL> else : <EOL> raise DBSecurityGroupNotFoundError ( security_group_name ) <EOL> return self . security_groups . values ( ) <EOL> def delete_security_group ( self , security_group_name ) : <EOL> if security_group_name in self . security_groups : <EOL> return self . security_groups . pop ( security_group_name ) <EOL> else : <EOL> raise DBSecurityGroupNotFoundError ( security_group_name ) <EOL> def authorize_security_group ( self , security_group_name , cidr_ip ) : <EOL> security_group = self . describe_security_groups ( security_group_name ) [ <NUM_LIT:0> ] <EOL> security_group . authorize_cidr ( cidr_ip ) <EOL> return security_group <EOL> def create_subnet_group ( self , subnet_name , description , subnets ) : <EOL> subnet_group = SubnetGroup ( subnet_name , description , subnets ) <EOL> self . subnet_groups [ subnet_name ] = subnet_group <EOL> return subnet_group <EOL> def describe_subnet_groups ( self , subnet_group_name ) : <EOL> if subnet_group_name : <EOL> if subnet_group_name in self . subnet_groups : <EOL> return [ self . subnet_groups [ subnet_group_name ] ] <EOL> else : <EOL> raise DBSubnetGroupNotFoundError ( subnet_group_name ) <EOL> return self . subnet_groups . values ( ) <EOL> def delete_subnet_group ( self , subnet_name ) : <EOL> if subnet_name in self . subnet_groups : <EOL> return self . subnet_groups . pop ( subnet_name ) <EOL> else : <EOL> raise DBSubnetGroupNotFoundError ( subnet_name ) <EOL> rds_backends = { } <EOL> for region in boto . rds . regions ( ) : <EOL> rds_backends [ region . name ] = RDSBackend ( ) </s>
