utils/test/testapi/opnfv_testapi/db/api.py

   1 import motor
   2
   3 from opnfv_testapi.common.config import CONF
   4
   5 DB = motor.MotorClient(CONF.mongo_url)[CONF.mongo_dbname]
   6
   7
   8 def db_update(collection, query, update_req):
   9     return _eval_db(collection, 'update', query, update_req, check_keys=False)
  10
  11
  12 def db_delete(collection, query):
  13     return _eval_db(collection, 'remove', query)
  14
  15
  16 def db_aggregate(collection, pipelines):
  17     return _eval_db(collection, 'aggregate', pipelines, allowDiskUse=True)
  18
  19
  20 def db_list(collection, query):
  21     return _eval_db(collection, 'find', query)
  22
  23
  24 def db_save(collection, data):
  25     return _eval_db(collection, 'insert', data, check_keys=False)
  26
  27
  28 def db_find_one(collection, query):
  29     return _eval_db(collection, 'find_one', query)
  30
  31
  32 def _eval_db(collection, method, *args, **kwargs):
  33     exec_collection = DB.__getattr__(collection)
  34     return exec_collection.__getattribute__(method)(*args, **kwargs)
  35
  36
  37 def _eval_db_find_one(query, table=None):
  38     return _eval_db(table, 'find_one', query)