def testing_engine(url=None, options=None):
 from sqlalchemy import create_engine
 from sqlalchemy.engine.url import make_url
 if (not options):
  use_reaper = True
 else:
  use_reaper = options.pop('use_reaper', True)
 url = (url or config.db.url)
 url = make_url(url)
 if (options is None):
  if ((config.db is None) or (url.drivername == config.db.url.drivername)):
   options = config.db_opts
  else:
   options = {}
 engine = create_engine(url, **options)
 engine._has_events = True
 if isinstance(engine.pool, pool.QueuePool):
  engine.pool._timeout = 0
  engine.pool._max_overflow = 0
 if use_reaper:
  event.listen(engine.pool, 'connect', testing_reaper.connect)
  event.listen(engine.pool, 'checkout', testing_reaper.checkout)
  event.listen(engine.pool, 'invalidate', testing_reaper.invalidate)
  testing_reaper.add_engine(engine)
 return engine

def testing_engine ( url = None , options = None ) : from sqlalchemy import create_engine from sqlalchemy . engine . url import make_url if ( not options ) : use_reaper = True else : use_reaper = options . pop ( 'use_reaper' , True ) url = ( url or config . db . url ) url = make_url ( url ) if ( options is None ) : if ( ( config . db is None ) or ( url . drivername == config . db . url . drivername ) ) : options = config . db_opts else : options = { } engine = create_engine ( url , ** options ) engine . _has_events = True if isinstance ( engine . pool , pool . QueuePool ) : engine . pool . _timeout = 0 engine . pool . _max_overflow = 0 if use_reaper : event . listen ( engine . pool , 'connect' , testing_reaper . connect ) event . listen ( engine . pool , 'checkout' , testing_reaper . checkout ) event . listen ( engine . pool , 'invalidate' , testing_reaper . invalidate ) testing_reaper . add_engine ( engine ) return engine
produce an engine configured by --options with optional overrides .