ftests_db_in_memory.DBI Class Reference

Common use wrapper class to interact with the ORM databases Provide CRUD interface (create, read, update, delete) entries. More...

Inheritance diagram for ftests_db_in_memory.DBI:

Collaboration diagram for ftests_db_in_memory.DBI:

Public Member Functions
def	__init__ (self, config_dic)
	constructor initialize fields More...
def	getErrorCode (self)
	get error code return error code last performed action (e.g. More...
def	getErrorMsg (self)
	get error message return error message last performed action (e.g. More...
def	insert (self, task)
	insert insert objects into the database More...
def	fetch (self, obj, clause)
	fetch fetch objects from the database More...
def	sql (self, obj, clause)
	sql fetch objects from the database More...
def	fetchAll (self, tasks)
	fetchAll fetch all objects from the database More...
def	update (self, obj, clause)
	update update objects in the database More...
def	insertOnUpdate (self, tasks, clause)
	insertOnUpdate insert new object or update object if exists in the table More...
def	delete (self, obj, clause)
	delete delete objects from the database More...
def	deleteAll (self, tasks)
	deleteAll delete all objects from the database More...

Public Attributes
	session
	errorCode
	errorMessage

Detailed Description

Common use wrapper class to interact with the ORM databases Provide CRUD interface (create, read, update, delete) entries.

Definition at line 29 of file ftests_db_in_memory.py.

Constructor & Destructor Documentation

__init__()

def ftests_db_in_memory.DBI.__init__	(		self,
			config_dic
	)

constructor initialize fields

Parameters

config_dic

dict object instance from config section

Definition at line 37 of file ftests_db_in_memory.py.

  def __init__(self, config_dic):
    db_name = config_dic["db_name"]
    """
    db_name - database name
    Examples:
    db_name=sqlite://               # in memory
    db_name=sqlite:///:memory:      # in memory
    db_name=sqlite:///del.db        # relative path to file
    db_name=sqlite:////tmp/del.db   # absolute path to file
    """
    
    engine = create_engine('sqlite:///:memory:', echo=True)
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    self.session = Session()
    
    self.errorCode = 0
    self.errorMessage = "Ok"

Member Function Documentation

delete()

def ftests_db_in_memory.DBI.delete	(		self,
			obj,
			clause
	)

delete delete objects from the database

Parameters

obj-	list of objects (not mandatory the same type) for deleting
clause	- sql clause for deleting the object from the table

Returns

list of deleted objects can be checked by the getError() method

Definition at line 215 of file ftests_db_in_memory.py.

  def delete(self, obj, clause):
    self.errorCode = 0
    deleted = []
    try:
      deleted_task = db.session.query(type(obj)).filter(clause).first()
      if deleted_task:
        deleted.append(deleted_task)
        db.session.delete(deleted_task)
      db.session.commit()
    except:
      db.session.rollback()
      self.errorCode = 6
      self.errorMessage  = "delete error"
    return deleted

Here is the caller graph for this function:

deleteAll()

def ftests_db_in_memory.DBI.deleteAll	(		self,
			tasks
	)

deleteAll delete all objects from the database

Parameters

tasks

- object type for deleting

Returns

list of deleted objects can be checked by the getError() method

Definition at line 237 of file ftests_db_in_memory.py.

  def deleteAll(self, tasks):
    deleted = []
    self.errorCode = 0
    try:
      conn = db.engine.connect()
      if hasattr(tasks, "__len__"):
        for task in tasks:
          conn.execute(type(task).delete())
      else:
        #conn.execute(type(tasks).delete())
        #db.session.query(delete(tasks))
        #db.session.delete(tasks)
        ress = db.session.query(tasks).filter(tasks.UThreads=="").first()
        #ress = conn.execute(select([type(tasks)])).fetchall()
        for res in ress:
          db.session.delete(res)
      db.session.commit()
    except:
      db.session.rollback()
      self.errorCode = 7
      self.errorMessage  = "delete all error"
    return deleted

Here is the call graph for this function:

fetch()

def ftests_db_in_memory.DBI.fetch	(		self,
			obj,
			clause
	)

fetch fetch objects from the database

Parameters

obj	- list of objects (not mandatory the same type) for fetching
clause	- sql clause for fetching the object from table

Returns

list of fetched objects can be checked by the getError() method

Definition at line 102 of file ftests_db_in_memory.py.

  def fetch(self, obj, clause):
    self.errorCode = 0
    fetched = []
    try:
      fetched.append(db.session.query(type(obj)).filter(clause).first())
      #fetched.append(db.session.query(type(obj)).from_statement(clause).first())
    except:
      db.session.rollback()
      self.errorCode = 1
      self.errorMessage  = "Fetch error"
    return fetched

fetchAll()

def ftests_db_in_memory.DBI.fetchAll	(		self,
			tasks
	)

fetchAll fetch all objects from the database

Parameters

tasks

- object type for fetching

Returns

list of fetched objects can be checked by the getError() method

Definition at line 142 of file ftests_db_in_memory.py.

  def fetchAll(self, tasks):
    self.errorCode = 0
    fetched = []
    try:
      conn = db.engine.connect()
      if hasattr(tasks, "__len__"):
        for task in tasks:
          fetched.append(conn.execute(select([type(task)])).fetchall())
      else:
        fetched.append(conn.execute(select([type(tasks)])).fetchall())
    except:
      db.session.rollback()
      self.errorCode = 3
      self.errorMessage  = "fetch all error"
    return fetched

getErrorCode()

def ftests_db_in_memory.DBI.getErrorCode

(

self

)

get error code return error code last performed action (e.g.

insert, update, etc.)

Returns

error code last performed action (e.g. insert, update, etc.)

Definition at line 62 of file ftests_db_in_memory.py.

  def getErrorCode(self):
    return self.errorCode 

getErrorMsg()

def ftests_db_in_memory.DBI.getErrorMsg

(

self

)

get error message return error message last performed action (e.g.

insert, update, etc.)

Returns

error message last performed action (e.g. insert, update, etc.)

Definition at line 71 of file ftests_db_in_memory.py.

  def getErrorMsg(self):
    return self.errorMessage

insert()

def ftests_db_in_memory.DBI.insert	(		self,
			task
	)

insert insert objects into the database

Parameters

tasks

- list of objects (not mandatory the same type) for insertion

Returns

the same list of objects can be checked by the getError() method

Definition at line 82 of file ftests_db_in_memory.py.

  def insert(self, task):
    self.errorCode = 0
    self.session.add(task)
    #try:
    self.session.commit()
    #except:
      #db.session.rollback()
      #self.errorCode = CONSTANTS.DBI_INSERT_ERROR_CODE
      #self.errorMessage  = CONSTANTS.DBI_INSERT_ERROR_MSG
    return task

Here is the caller graph for this function:

insertOnUpdate()

def ftests_db_in_memory.DBI.insertOnUpdate	(		self,
			tasks,
			clause
	)

insertOnUpdate insert new object or update object if exists in the table

Parameters

tasks	- list of objects (not mandatory the same type) for updating
clause	- (deprecated) sql clause for updating the object in the table

Returns

list of updated objects can be checked by the getError() method

Definition at line 192 of file ftests_db_in_memory.py.

  def insertOnUpdate(self, tasks, clause):
    self.errorCode = 0
    try:
      if hasattr(tasks, "__len__"):
        for task in tasks:
          db.session.merge(task)
      else:
        db.session.merge(tasks)
      db.session.commit()
    except:
      db.session.rollback()
      self.errorCode = 5
      self.errorMessage  = "insert on update error"
    return tasks

sql()

def ftests_db_in_memory.DBI.sql	(		self,
			obj,
			clause
	)

sql fetch objects from the database

Parameters

obj	- list of objects (not mandatory the same type) for fetching
clause	- sql clause for fetching the object from table

Returns

list of fetched objects can be checked by the getError() method

Definition at line 123 of file ftests_db_in_memory.py.

  def sql(self, obj, clause):
    self.errorCode = 0
    fetched = []
    try:
      fetched.append(db.session.query(type(obj)).from_statement(clause).first())
    except:
      db.session.rollback()
      self.errorCode = 2 
      self.errorMessage  = "sql error"
    return fetched

update()

def ftests_db_in_memory.DBI.update	(		self,
			obj,
			clause
	)

update update objects in the database

Parameters

tasks	- list of objects (not mandatory the same type) for updating
clause	- sql clause for updating the object in the table

Returns

list of updated objects can be checked by the getError() method

Definition at line 167 of file ftests_db_in_memory.py.

  def update(self, obj, clause):
    self.errorCode = 0
    updated = []
    try:
      updated_task = db.session.query(type(obj)).filter(clause).first()
      if updated_task:
        attributes = [attr for attr in dir(obj) if not attr.startswith('__') and not attr.startswith('_') and getattr(obj,attr)]
        for attr in attributes:
          setattr(updated_task, attr, getattr(obj, attr))
          updated.append(updated_task)
      db.session.commit()
    except:
      db.session.rollback()
      self.errorCode = 4
      self.errorMessage  = "update error"
    return updated

Member Data Documentation

errorCode

ftests_db_in_memory.DBI.errorCode

Definition at line 53 of file ftests_db_in_memory.py.

errorMessage

ftests_db_in_memory.DBI.errorMessage

Definition at line 54 of file ftests_db_in_memory.py.

session

ftests_db_in_memory.DBI.session

Definition at line 51 of file ftests_db_in_memory.py.

---

The documentation for this class was generated from the following file:

• sources/hce/ftests/dbi/ftests_db_in_memory.py