Inheritance diagram for o2o_db_manager.DbManager:

Public Member Functions
def	__init__ (self, db, authPath=None)

def	check_table (self, table_def, table_class)

def	connect (self, url=None)

def	get_url (self, force_schema=False)

Public Attributes
	authPath

	db_service

	engine

	schema

	session

Detailed Description

Definition at line 29 of file o2o_db_manager.py.

Constructor & Destructor Documentation

◆ init()

def o2o_db_manager.DbManager.__init__	(	self,
		db,
		authPath = `None`
	)

Definition at line 30 of file o2o_db_manager.py.

     def __init__(self, db, authPath=None):
         self.authPath = authPath
         if db == 'prod':
             self.db_service = prod_db_service
         elif db == 'dev':
             self.db_service = dev_db_service
         elif db == 'private':
             self.db_service = None
         else:
             raise RuntimeError('Option db(=%s) is not in the supported database list: [prod, dev, private]' % db)
  
         logging.info('Connecting to %s database' % self.db_service[0] if self.db_service else private_db)
  
         self.schema = schema_dict[self.db_service[0]] if self.db_service else None
  

Member Function Documentation

◆ check_table()

def o2o_db_manager.DbManager.check_table	(	self,
		table_def,
		table_class
	)

Definition at line 56 of file o2o_db_manager.py.

     def check_table(self, table_def, table_class):
         self.engine = sqlalchemy.create_engine(self.get_url())
         if not self.engine.has_table(table_def.__tablename__, self.schema):
             logging.info('Creating table %s on %s' % (table_def.__tablename__,
                                                       self.db_service[0] if self.db_service else private_db))
             self.engine = sqlalchemy.create_engine(self.get_url(True))
             table_class.__table__.create(bind=self.engine)
         self.session = sqlalchemy.orm.scoped_session(sqlalchemy.orm.sessionmaker(bind=self.engine))
  

Referenced by o2o_db_cfgmap.DbManagerDAQ.update_hashmap().

◆ connect()

def o2o_db_manager.DbManager.connect	(	self,
		url = `None`
	)

Definition at line 65 of file o2o_db_manager.py.

     def connect(self, url=None):
         engine = sqlalchemy.create_engine(url) if url else self.engine
         session = sqlalchemy.orm.scoped_session(sqlalchemy.orm.sessionmaker(bind=engine))
         return session
  

References o2o_db_manager.DbManager.engine, querying.connection.engine, and conddblib.Connection.engine.

Referenced by o2o_db_cfgmap.DbManagerDAQ.update_hashmap().

◆ get_url()

def o2o_db_manager.DbManager.get_url	(	self,
		force_schema = `False`
	)

Definition at line 45 of file o2o_db_manager.py.

     def get_url(self, force_schema=False):
         if self.db_service == None:
             url = private_db
         else:
             authEntry = self.db_service[1]
             if force_schema and self.schema:
                 authEntry = '%s/%s' % (self.db_service[0], self.schema)
             username, _, pwd = auth.get_credentials( authEntry, self.authPath)
             url = sqlalchemy_tpl % (username, pwd, self.db_service[0])
         return url
  