dbi_demo.DemoTaskManager Class Reference

Inheritance diagram for dbi_demo.DemoTaskManager:

Collaboration diagram for dbi_demo.DemoTaskManager:

Public Member Functions
def	__init__ (self, config_dic)
def	process (self, input_tasks)

Public Attributes
	dbi

Detailed Description

Definition at line 25 of file dbi_demo.py.

Constructor & Destructor Documentation

__init__()

def dbi_demo.DemoTaskManager.__init__	(		self,
			config_dic
	)

Definition at line 28 of file dbi_demo.py.

  def __init__(self, config_dic):
    # create dbi instance
    self.dbi = DBI(config_dic)

Member Function Documentation

process()

def dbi_demo.DemoTaskManager.process	(		self,
			input_tasks
	)

Definition at line 33 of file dbi_demo.py.

  def process(self, input_tasks):
    t1 = TasksDataTable()
    t2 = TasksDataTable()
    t1.id = 11
    t2.id = 11
    t1.data = "fake data 1"
    t2.data = "fake data 2"
    self.dbi.insert(t1)
    print self.dbi.getErrorCode()
    self.dbi.insert(t2)
    print self.dbi.getErrorCode()
    #print self.dbi.fetch(t1, "id=%s"%(t1.id))
    #print db.session.query(type(t1)).filter(t1.id=t1.id, t1.data=t1.data)
    print db.session.query(type(t1)).filter_by(id=t1.id, data=t1.data).all()
    #print self.dbi.sql(t1, "select * from tasks_data_table where id=%s and data='%s'"%(t1.id,t1.data))
    print self.dbi.getErrorCode()
    return
    """
    # insert one object
    input_back_log_one_task = DemoBackLogTask(input_tasks[0])
    self.dbi.insert(input_back_log_one_task)
    if self.dbi.getErrorCode() != CONSTANTS.DBI_SUCCESS_CODE:
      # handle insertion's error 
      print "One insert error!"
      return
    else:
      # do ptocessing after insertion
      pass
    """
    """
    # insert array of objects
    input_back_log_tasks = [DemoBackLogTask(input_task) for input_task in input_tasks]
    self.dbi.insert(input_back_log_tasks)
    if self.dbi.getErrorCode() != CONSTANTS.DBI_SUCCESS_CODE:
      # handle insertion's error 
      print "Bunch insert error!"
      return
    else:
      # do ptocessing after insertion
      pass
    """
    return
    # fetch one object
    fetch_back_log_tasks = DemoBackLogTask(input_tasks[0])
    fetched_back_log_tasks = self.dbi.fetch(fetch_back_log_tasks, "id=%s"%fetch_back_log_tasks.id)
    if self.dbi.getErrorCode() != CONSTANTS.DBI_SUCCESS_CODE:
      # handle fetch's error 
        print "One fetch error!"
        return
    else:
      # do ptocessing after insertion
      pass
      for task in fetched_back_log_tasks:
        print task

    """
    # fetch bunch of objects
    fetch_back_log_tasks = [DemoBackLogTask(input_task) for input_task in input_tasks]
    fetched_back_log_tasks = self.dbi.fetch(fetch_back_log_tasks)
    if self.dbi.getErrorCode() != CONSTANTS.DBI_SUCCESS_CODE:
      # handle fetch's error 
      print "Bunch of fetch error!"
      return
    else:
      # do ptocessing after insertion
      pass
      for task in fetched_back_log_tasks:
        print task
    """

    """
    # fetch all for one object
    fetch_back_log_tasks = DemoBackLogTask(input_tasks[0])
    fetched_back_log_tasks = self.dbi.fetchAll(fetch_back_log_tasks)
    if self.dbi.getErrorCode() != CONSTANTS.DBI_SUCCESS_CODE:
      # handle fetch's error 
      print "fetch all for one object error!"
      return
    else:
      # do ptocessing after insertion
      pass
      for task in fetched_back_log_tasks:
        print task
        pass
    """

    """
    # fetch all for bunch of objects
    fetch_back_log_tasks = [DemoBackLogTask(input_task) for input_task in input_tasks]
    fetched_back_log_tasks = self.dbi.fetchAll(fetch_back_log_tasks)
    if self.dbi.getErrorCode() != CONSTANTS.DBI_SUCCESS_CODE:
      # handle fetch's error 
      print "fetch all for bunch of objects error!"
      return
    else:
      # do ptocessing after insertion
      pass
      for task in fetched_back_log_tasks:
        print task
        pass
    """

    """
    # update one object
    update_back_log_tasks = DemoBackLogTask(Task(id="11111", Tries=22))
    updated_back_log_tasks = self.dbi.update(update_back_log_tasks, "id=%s"%update_back_log_tasks.id)
    if self.dbi.getErrorCode() != CONSTANTS.DBI_SUCCESS_CODE:
      # handle update's error 
        print "update one object error!"
        return
    else:
    # do ptocessing after insertion
      pass
      for task in updated_back_log_tasks:
        #print vars(task)
        pass
    """

    """
    # update bunch of objects
    update_back_log_tasks = [DemoBackLogTask(input_task) for input_task in input_tasks]
    updated_back_log_tasks = self.dbi.update(update_back_log_tasks)
    if self.dbi.getErrorCode() != CONSTANTS.DBI_SUCCESS_CODE:
      # handle update's error 
      print "update bunch of objects error!"
      return
    else:
      # do ptocessing after insertion
      pass
      for task in updated_back_log_tasks:
        #print vars(task)
        pass
    """

    """
    # delete one object
    delete_back_log_tasks = DemoBackLogTask(Task(id="11111"))
    deleted_back_log_tasks = self.dbi.delete(delete_back_log_tasks, "id=%s"%delete_back_log_tasks.id)
    if self.dbi.getErrorCode() != CONSTANTS.DBI_SUCCESS_CODE:
      # handle update's error 
      print "delete one object error!"
      return
    else:
      # do ptocessing after insertion
      pass
      for task in deleted_back_log_tasks:
        print vars(task)
    """

    """
    # delete bunch of objects
    delete_back_log_tasks = [DemoBackLogTask(input_task) for input_task in input_tasks]
    deleted_back_log_tasks = self.dbi.delete(delete_back_log_tasks)
    if self.dbi.getErrorCode() != CONSTANTS.DBI_SUCCESS_CODE:
      # handle update's error 
      print "delete bunch of objects error!"
      return
    else:
      # do ptocessing after insertion
      pass
      for task in deleted_back_log_tasks:
        print vars(task)
    """


    # delete all for one object
    delete_back_log_tasks = DemoBackLogTask(input_tasks[0])
    deleted_back_log_tasks = self.dbi.deleteAll(delete_back_log_tasks)
    if self.dbi.getErrorCode() != CONSTANTS.DBI_SUCCESS_CODE:
      # handle update's error 
      print "delete all for one object error!"
      return
    else:
      # do ptocessing after insertion
      pass
      for task in deleted_back_log_tasks:
        print vars(task)


    """
    # delete all for bunch of objects
    delete_back_log_tasks = [DemoBackLogTask(input_task) for input_task in input_tasks]
    deleted_back_log_tasks = self.dbi.deleteAll(delete_back_log_tasks)
    if self.dbi.getErrorCode() != CONSTANTS.DBI_SUCCESS_CODE:
      # handle update's error 
      print "delete all for bunch of objects error!"
      return
    else:
      # do ptocessing after insertion
      pass
      for task in deleted_back_log_tasks:
        print vars(task)
    """

Here is the caller graph for this function:

Member Data Documentation

dbi

dbi_demo.DemoTaskManager.dbi

Definition at line 30 of file dbi_demo.py.

---

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

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