<s> """<STR_LIT>""" <EOL> import MySQLdb <EOL> import MySQLdb . cursors <EOL> import psycopg2 <EOL> from . . config import Config <EOL> from . . utils . exceptions import ETLConfigError <EOL> from . . utils . helpers import exactly_one <EOL> from . . utils . helpers import retry <EOL> from . . utils . hook import hook <EOL> config = Config ( ) <EOL> CONNECTION_RETRIES = config . etl . get ( '<STR_LIT>' , <NUM_LIT:2> ) <EOL> def get_redshift_config ( ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( config , '<STR_LIT>' ) : <EOL> raise ETLConfigError ( '<STR_LIT>' ) <EOL> return config . redshift <EOL> @ retry ( CONNECTION_RETRIES , <NUM_LIT> ) <EOL> @ hook ( '<STR_LIT>' ) <EOL> def redshift_connection ( redshift_creds = None , autocommit = True , <EOL> connect_timeout = <NUM_LIT:30> , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if redshift_creds is None : <EOL> redshift_creds = get_redshift_config ( ) <EOL> connection = psycopg2 . connect ( <EOL> host = redshift_creds [ '<STR_LIT>' ] , <EOL> user = redshift_creds [ '<STR_LIT>' ] , <EOL> password = redshift_creds [ '<STR_LIT>' ] , <EOL> port = redshift_creds [ '<STR_LIT>' ] , <EOL> database = redshift_creds [ '<STR_LIT>' ] , <EOL> connect_timeout = connect_timeout , <EOL> ** kwargs ) <EOL> connection . autocommit = autocommit <EOL> return connection <EOL> def get_sql_config ( database_name ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( config , '<STR_LIT>' ) : <EOL> raise ETLConfigError ( '<STR_LIT>' ) <EOL> if database_name not in config . mysql : <EOL> raise ETLConfigError ( <EOL> '<STR_LIT>' % database_name ) <EOL> sql_creds = config . mysql [ database_name ] <EOL> sql_creds [ '<STR_LIT>' ] = database_name <EOL> return sql_creds <EOL> @ retry ( CONNECTION_RETRIES , <NUM_LIT> ) <EOL> @ hook ( '<STR_LIT>' ) <EOL> def rds_connection ( database_name = None , sql_creds = None , <EOL> cursorclass = MySQLdb . cursors . SSCursor , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> assert exactly_one ( database_name , sql_creds ) , '<STR_LIT>' <EOL> if sql_creds is None : <EOL> sql_creds = get_sql_config ( database_name ) <EOL> connection = MySQLdb . connect ( <EOL> host = sql_creds [ '<STR_LIT>' ] , <EOL> user = sql_creds [ '<STR_LIT>' ] , <EOL> passwd = sql_creds [ '<STR_LIT>' ] , <EOL> db = sql_creds [ '<STR_LIT>' ] , <EOL> charset = '<STR_LIT:utf8>' , <EOL> cursorclass = cursorclass , <EOL> ** kwargs ) <EOL> return connection <EOL> def get_postgres_config ( ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( config , '<STR_LIT>' ) : <EOL> raise ETLConfigError ( '<STR_LIT>' ) <EOL> return config . postgres <EOL> @ retry ( CONNECTION_RETRIES , <NUM_LIT> ) <EOL> @ hook ( '<STR_LIT>' ) <EOL> def postgres_connection ( postgres_creds = None , autocommit = True , <EOL> connect_timeout = <NUM_LIT:30> , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if postgres_creds is None : <EOL> postgres_creds = get_postgres_config ( ) <EOL> connection = psycopg2 . connect ( <EOL> rds_instance = postgres_creds [ '<STR_LIT>' ] , <EOL> user = postgres_creds [ '<STR_LIT>' ] , <EOL> password = postgres_creds [ '<STR_LIT>' ] , <EOL> region = postgres_creds [ '<STR_LIT>' ] , <EOL> database = postgres_creds [ '<STR_LIT>' ] , <EOL> connect_timeout = connect_timeout , <EOL> ** kwargs ) <EOL> connection . autocommit = autocommit <EOL> return connection </s>
