<s> """<STR_LIT>""" <EOL> import json <EOL> from collections import defaultdict <EOL> from . pipeline_object import PipelineObject <EOL> from . utils import list_pipeline_instances <EOL> from . utils import get_datapipeline_connection <EOL> from . . utils . exceptions import ETLInputError <EOL> class DataPipeline ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , unique_id = None , name = None , pipeline_id = None , <EOL> tags = None , description = None ) : <EOL> """<STR_LIT>""" <EOL> self . conn = get_datapipeline_connection ( ) <EOL> self . objects = [ ] <EOL> if pipeline_id : <EOL> if unique_id or name : <EOL> raise ETLInputError ( '<STR_LIT>' ) <EOL> self . pipeline_id = pipeline_id <EOL> else : <EOL> if unique_id is None : <EOL> raise ETLInputError ( '<STR_LIT>' ) <EOL> if not name : <EOL> name = unique_id <EOL> response = self . custom_create_pipeline ( <EOL> name , unique_id , description , tags ) <EOL> self . pipeline_id = response [ '<STR_LIT>' ] <EOL> @ property <EOL> def id ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . pipeline_id <EOL> @ property <EOL> def aws_format ( self ) : <EOL> """<STR_LIT>""" <EOL> return [ x . aws_format ( ) for x in self . objects ] <EOL> def add_object ( self , pipeline_object ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( pipeline_object , PipelineObject ) : <EOL> raise ETLInputError ( <EOL> '<STR_LIT>' ) <EOL> self . objects . append ( pipeline_object ) <EOL> def validate_pipeline_definition ( self ) : <EOL> """<STR_LIT>""" <EOL> response = self . conn . validate_pipeline_definition ( <EOL> self . aws_format , self . id ) <EOL> return response . get ( '<STR_LIT>' , None ) <EOL> def update_pipeline_definition ( self ) : <EOL> """<STR_LIT>""" <EOL> self . conn . put_pipeline_definition ( self . aws_format , self . id ) <EOL> def activate ( self ) : <EOL> """<STR_LIT>""" <EOL> self . conn . activate_pipeline ( self . id ) <EOL> def delete ( self ) : <EOL> """<STR_LIT>""" <EOL> self . conn . delete_pipeline ( self . pipeline_id ) <EOL> def instance_details ( self ) : <EOL> """<STR_LIT>""" <EOL> instances = list_pipeline_instances ( self . pipeline_id , self . conn ) <EOL> result = defaultdict ( list ) <EOL> for instance in instances : <EOL> result [ instance [ '<STR_LIT>' ] ] . append ( instance ) <EOL> return result <EOL> def custom_create_pipeline ( self , name , unique_id , description = None , <EOL> tags = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT:name>' : name , '<STR_LIT>' : unique_id , } <EOL> if description is not None : <EOL> params [ '<STR_LIT:description>' ] = description <EOL> if tags is not None : <EOL> params [ '<STR_LIT>' ] = tags <EOL> return self . conn . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) </s>
