<s> """<STR_LIT>""" <EOL> from sqlalchemy import create_engine , event <EOL> from sqlalchemy . exc import ArgumentError <EOL> from . . bases import BasePartitionFeature , BaseOperationFeature <EOL> from ... exceptions import OptionNotSetError , OptionValueError <EOL> class ConnectionMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> @ property <EOL> def connection ( self ) : <EOL> if self . model_cls . metadata . is_bound ( ) : <EOL> return self . model_cls . metadata . bind <EOL> try : <EOL> return create_engine ( self . options [ '<STR_LIT>' ] ) <EOL> except KeyError as key : <EOL> raise OptionNotSetError ( model = self . model_cls . __name__ , current = key ) <EOL> except ArgumentError as e : <EOL> raise OptionValueError ( model = self . model_cls . __name__ , current = self . options [ '<STR_LIT>' ] , option = '<STR_LIT>' , cause = e ) <EOL> class OperationFeature ( ConnectionMixin , BaseOperationFeature ) : <EOL> def execute ( self , sql , autocommit = True ) : <EOL> return self . connection . execution_options ( autocommit = autocommit ) . execute ( sql . replace ( '<STR_LIT:%>' , '<STR_LIT>' ) ) <EOL> class PartitionFeature ( ConnectionMixin , BasePartitionFeature ) : <EOL> @ property <EOL> def model_meta ( self ) : <EOL> return { <EOL> '<STR_LIT>' : self . model_cls . __table__ . name , <EOL> '<STR_LIT>' : self . model_cls . __table__ . primary_key . columns . keys ( ) , <EOL> '<STR_LIT>' : self . connection . dialect . name , <EOL> '<STR_LIT>' : self . _column_value ( self . model_cls . __table__ . columns . keys ( ) ) , <EOL> } <EOL> @ staticmethod <EOL> def register_hooks ( model ) : <EOL> """<STR_LIT>""" <EOL> def before_insert ( * args ) : <EOL> partition = args [ <NUM_LIT:2> ] . architect . partition . get_partition ( ) <EOL> if not partition . exists ( ) : <EOL> partition . create ( ) <EOL> event . listen ( model , '<STR_LIT>' , before_insert ) </s>
