<s> import subprocess <EOL> import click <EOL> from catwatch . lib . db_seed import seed_database <EOL> from catwatch . app import create_app <EOL> from catwatch . extensions import db <EOL> app = create_app ( ) <EOL> db . app = app <EOL> SQLALCHEMY_DATABASE_URI = app . config . get ( '<STR_LIT>' , None ) <EOL> class PostgresDatabase ( object ) : <EOL> def __init__ ( self , databases ) : <EOL> self . _config = _parse_database_uri ( SQLALCHEMY_DATABASE_URI ) <EOL> self . _psql = '<STR_LIT>' <EOL> self . _createdb = '<STR_LIT>' <EOL> self . _dropdb = '<STR_LIT>' <EOL> self . _container = _get_postgres_container ( ) <EOL> self . _docker_exec = '<STR_LIT>' . format ( self . _container ) <EOL> self . databases = databases <EOL> if databases == ( ) : <EOL> self . databases = [ self . _config . get ( '<STR_LIT>' ) ] <EOL> @ property <EOL> def config ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _config <EOL> @ property <EOL> def container ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _container <EOL> def psql ( self , command ) : <EOL> """<STR_LIT>""" <EOL> pg = '<STR_LIT>' . format ( self . _psql , command ) <EOL> shell_command = '<STR_LIT>' . format ( self . _docker_exec , pg ) <EOL> return subprocess . call ( shell_command , shell = True ) <EOL> def _user ( self ) : <EOL> """<STR_LIT>""" <EOL> create_role = "<STR_LIT>" . format ( <EOL> self . config . get ( '<STR_LIT:username>' ) , self . config . get ( '<STR_LIT:password>' ) ) <EOL> return self . psql ( create_role ) <EOL> def list ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . psql ( '<STR_LIT>' ) <EOL> def create ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _user ( ) <EOL> for database in self . databases : <EOL> pg = '<STR_LIT>' . format ( self . _createdb , database ) <EOL> command = '<STR_LIT>' . format ( self . _docker_exec , pg ) <EOL> subprocess . call ( command , shell = True ) <EOL> return None <EOL> def drop ( self ) : <EOL> """<STR_LIT>""" <EOL> for database in self . databases : <EOL> pg = '<STR_LIT>' . format ( self . _dropdb , database ) <EOL> command = '<STR_LIT>' . format ( self . _docker_exec , pg ) <EOL> subprocess . call ( command , shell = True ) <EOL> return None <EOL> def _parse_database_uri ( uri ) : <EOL> """<STR_LIT>""" <EOL> db_engine = { } <EOL> uri_parts = uri . split ( '<STR_LIT>' ) <EOL> db_engine [ '<STR_LIT>' ] = uri_parts [ <NUM_LIT:0> ] <EOL> uri_without_protocol_parts = uri_parts [ <NUM_LIT:1> ] . split ( '<STR_LIT:@>' ) <EOL> user_and_pass = uri_without_protocol_parts [ <NUM_LIT:0> ] . split ( '<STR_LIT::>' ) <EOL> domain_and_db = uri_without_protocol_parts [ <NUM_LIT:1> ] . split ( '<STR_LIT:/>' ) <EOL> db_engine [ '<STR_LIT:username>' ] = user_and_pass [ <NUM_LIT:0> ] <EOL> db_engine [ '<STR_LIT:password>' ] = user_and_pass [ <NUM_LIT:1> ] <EOL> db_engine [ '<STR_LIT:host>' ] = domain_and_db [ <NUM_LIT:0> ] . split ( '<STR_LIT::>' ) [ <NUM_LIT:0> ] <EOL> db_engine [ '<STR_LIT:port>' ] = domain_and_db [ <NUM_LIT:0> ] . split ( '<STR_LIT::>' ) [ <NUM_LIT:1> ] <EOL> db_engine [ '<STR_LIT>' ] = domain_and_db [ <NUM_LIT:1> ] <EOL> return db_engine <EOL> def _get_postgres_container ( ) : <EOL> """<STR_LIT>""" <EOL> find_container = '''<STR_LIT>''' <EOL> container_id = subprocess . check_output ( find_container , shell = True ) [ : - <NUM_LIT:1> ] <EOL> return container_id <EOL> @ click . group ( ) <EOL> def cli ( ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> @ click . command ( ) <EOL> def list ( ) : <EOL> """<STR_LIT>""" <EOL> database = PostgresDatabase ( ( ) ) <EOL> database . list ( ) <EOL> @ click . command ( ) <EOL> @ click . argument ( '<STR_LIT>' ) <EOL> def psql ( command ) : <EOL> """<STR_LIT>""" <EOL> database = PostgresDatabase ( ( ) ) <EOL> database . psql ( command ) <EOL> @ click . command ( ) <EOL> @ click . argument ( '<STR_LIT>' , nargs = - <NUM_LIT:1> ) <EOL> def create ( databases ) : <EOL> """<STR_LIT>""" <EOL> database = PostgresDatabase ( databases ) <EOL> database . create ( ) <EOL> return db . create_all ( ) <EOL> @ click . command ( ) <EOL> @ click . argument ( '<STR_LIT>' , nargs = - <NUM_LIT:1> ) <EOL> def drop ( databases ) : <EOL> """<STR_LIT>""" <EOL> database = PostgresDatabase ( databases ) <EOL> database . drop ( ) <EOL> return None <EOL> @ click . command ( ) <EOL> @ click . argument ( '<STR_LIT>' , nargs = - <NUM_LIT:1> ) <EOL> @ click . pass_context <EOL> def reset ( ctx , databases ) : <EOL> """<STR_LIT>""" <EOL> ctx . invoke ( drop , databases = databases ) <EOL> ctx . invoke ( create , databases = databases ) <EOL> ctx . invoke ( seed ) <EOL> return None <EOL> @ click . command ( ) <EOL> def init ( ) : <EOL> """<STR_LIT>""" <EOL> db . drop_all ( ) <EOL> return db . create_all ( ) <EOL> @ click . command ( ) <EOL> def seed ( ) : <EOL> """<STR_LIT>""" <EOL> return seed_database ( ) <EOL> cli . add_command ( list ) <EOL> cli . add_command ( psql ) <EOL> cli . add_command ( create ) <EOL> cli . add_command ( drop ) <EOL> cli . add_command ( reset ) <EOL> cli . add_command ( init ) <EOL> cli . add_command ( seed ) </s>
