No Description
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

datasource.py 39KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890
  1. #
  2. # This file is part of Lodel 2 (https://github.com/OpenEdition)
  3. #
  4. # Copyright (C) 2015-2017 Cléo UMS-3287
  5. #
  6. # This program is free software: you can redistribute it and/or modify
  7. # it under the terms of the GNU Affero General Public License as published
  8. # by the Free Software Foundation, either version 3 of the License, or
  9. # (at your option) any later version.
  10. #
  11. # This program is distributed in the hope that it will be useful,
  12. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  14. # GNU Affero General Public License for more details.
  15. #
  16. # You should have received a copy of the GNU Affero General Public License
  17. # along with this program. If not, see <http://www.gnu.org/licenses/>.
  18. #
  19. ## @package plugins.mongodb_datasource.datasource Main datasource module
  20. #
  21. # In this class, there is the MongoDbDatasource class, that handles the basic
  22. # operations that can be done (CRUD ones).
  23. import re
  24. import warnings
  25. import copy
  26. import functools
  27. from bson.son import SON
  28. from collections import OrderedDict
  29. import pymongo
  30. from pymongo.errors import BulkWriteError
  31. from lodel.context import LodelContext
  32. LodelContext.expose_modules(globals(), {
  33. 'lodel.logger': 'logger',
  34. 'lodel.leapi.leobject': ['CLASS_ID_FIELDNAME'],
  35. 'lodel.leapi.datahandlers.base_classes': ['Reference', 'MultipleRef'],
  36. 'lodel.exceptions': ['LodelException', 'LodelFatalError'],
  37. 'lodel.plugin.datasource_plugin': ['AbstractDatasource']})
  38. from . import utils
  39. from .exceptions import *
  40. from .utils import object_collection_name, collection_name, \
  41. MONGODB_SORT_OPERATORS_MAP, connection_string, mongo_fieldname
  42. ## @brief Datasource class
  43. # @ingroup plugin_mongodb_datasource
  44. class MongoDbDatasource(AbstractDatasource):
  45. ## @brief Stores existing connections
  46. #
  47. # The key of this dict is a hash built upon the connection string and the
  48. # ro (read-only) parameter.
  49. #
  50. # The value is a dict with 2 keys :
  51. # - conn_count : the number of instanciated datasource that use this
  52. # connection
  53. # - db : the pymongo database object instance
  54. _connections = dict()
  55. ## @brief Mapping from lodel2 operators to mongodb operators
  56. lodel2mongo_op_map = {
  57. '=':'$eq', '<=':'$lte', '>=':'$gte', '!=':'$ne', '<':'$lt',
  58. '>':'$gt', 'in':'$in', 'not in':'$nin' }
  59. ## @brief List of mongodb operators that expect re as value
  60. mongo_op_re = ['$in', '$nin']
  61. wildcard_re = re.compile('[^\\\\]\*')
  62. ## @brief instanciates a database object given a connection name
  63. # @param host str : hostname or IP
  64. # @param port int : mongodb listening port
  65. # @param db_name str
  66. # @param username str
  67. # @param password str
  68. # @param read_only bool : If True the Datasource is for read only, else the
  69. # Datasource is write only !
  70. def __init__(self, host, port, db_name, username, password, read_only = False):
  71. ## @brief Connections infos that can be kept securly
  72. self.__db_infos = {'host': host, 'port': port, 'db_name': db_name}
  73. ## @brief Is the instance read only ? (if not it's write only)
  74. self.__read_only = bool(read_only)
  75. ## @brief Uniq ID for mongodb connection
  76. self.__conn_hash= None
  77. ## @brief Stores the database cursor
  78. self.database = self.__connect(
  79. username, password, db_name, self.__read_only)
  80. ## @brief Destructor that attempt to close connection to DB
  81. #
  82. # Decrease the conn_count of associated MongoDbDatasource::_connections
  83. # item. If it reach 0 close the connection to the db
  84. # @see MongoDbDatasource::__connect()
  85. def __del__(self):
  86. self._connections[self.__conn_hash]['conn_count'] -= 1
  87. if self._connections[self.__conn_hash]['conn_count'] <= 0:
  88. self._connections[self.__conn_hash]['db'].close()
  89. del(self._connections[self.__conn_hash])
  90. logger.info("Closing connection to database")
  91. ## @brief Provides a new uniq numeric ID
  92. # @param emcomp LeObject subclass (not instance) : To know on wich things we
  93. # have to be uniq
  94. # @warning multiple UID broken by this method
  95. # @return an integer
  96. def new_numeric_id(self, emcomp):
  97. target = emcomp.uid_source()
  98. tuid = target._uid[0] # Multiple UID broken here
  99. results = self.select(
  100. target, field_list = [tuid], filters = [],
  101. order=[(tuid, 'DESC')], limit = 1)
  102. if len(results) == 0:
  103. return 1
  104. return results[0][tuid]+1
  105. ## @brief returns a selection of documents from the datasource
  106. # @param target Emclass
  107. # @param field_list list
  108. # @param filters list : List of filters
  109. # @param relational_filters list : List of relational filters
  110. # @param order list : List of column to order. ex: order = [('title', 'ASC'),]
  111. # @param group list : List of tupple representing the column used as
  112. # "group by" fields. ex: group = [('title', 'ASC'),]
  113. # @param limit int : Number of records to be returned
  114. # @param offset int: used with limit to choose the start record
  115. # @return list
  116. # @todo Implement group for abstract LeObject childs
  117. def select(self, target, field_list, filters = None,
  118. relational_filters=None, order=None, group=None, limit=None,
  119. offset=0):
  120. if target.is_abstract():
  121. # Reccursive calls for abstract LeObject child
  122. results = self.__act_on_abstract(target, filters,
  123. relational_filters, self.select, field_list = field_list,
  124. order = order, group = group, limit = limit)
  125. # Here we may implement the group
  126. # If sorted query we have to sort again
  127. if order is not None:
  128. key_fun = functools.cmp_to_key(
  129. self.__generate_lambda_cmp_order(order))
  130. results = sorted(results, key=key_fun)
  131. # If limit given apply limit again
  132. if offset > len(results):
  133. results = list()
  134. else:
  135. if limit is not None:
  136. if limit + offset > len(results):
  137. limit = len(results)-offset-1
  138. results = results[offset:offset+limit]
  139. return results
  140. # Default behavior
  141. if filters is None:
  142. filters = list()
  143. if relational_filters is None:
  144. relational_filters = list()
  145. collection_name = object_collection_name(target)
  146. collection = self.database[collection_name]
  147. query_filters = self.__process_filters(
  148. target, filters, relational_filters)
  149. query_result_ordering = None
  150. if order is not None:
  151. query_result_ordering = utils.parse_query_order(order)
  152. if group is None:
  153. if field_list is None:
  154. field_list = dict()
  155. else:
  156. f_list=dict()
  157. for fl in field_list:
  158. f_list[fl] = 1
  159. field_list = f_list
  160. field_list['_id'] = 0
  161. cursor = collection.find(
  162. spec = query_filters,
  163. fields=field_list,
  164. skip=offset,
  165. limit=limit if limit != None else 0,
  166. sort=query_result_ordering)
  167. else:
  168. pipeline = list()
  169. unwinding_list = list()
  170. grouping_dict = OrderedDict()
  171. sorting_list = list()
  172. for group_param in group:
  173. field_name = group_param[0]
  174. field_sort_option = group_param[1]
  175. sort_option = MONGODB_SORT_OPERATORS_MAP[field_sort_option]
  176. unwinding_list.append({'$unwind': '$%s' % field_name})
  177. grouping_dict[field_name] = '$%s' % field_name
  178. sorting_list.append((field_name, sort_option))
  179. sorting_list.extends(query_result_ordering)
  180. pipeline.append({'$match': query_filters})
  181. if field_list is not None:
  182. pipeline.append({
  183. '$project': SON([{field_name: 1}
  184. for field_name in field_list])})
  185. pipeline.extend(unwinding_list)
  186. pipeline.append({'$group': grouping_dict})
  187. pipeline.extend({'$sort': SON(sorting_list)})
  188. if offset > 0:
  189. pipeline.append({'$skip': offset})
  190. if limit is not None:
  191. pipeline.append({'$limit': limit})
  192. results = list()
  193. for document in cursor:
  194. results.append(document)
  195. return results
  196. ## @brief Deletes records according to given filters
  197. # @param target Emclass : class of the record to delete
  198. # @param filters list : List of filters
  199. # @param relational_filters list : List of relational filters
  200. # @return int : number of deleted records
  201. def delete(self, target, filters, relational_filters):
  202. if target.is_abstract():
  203. logger.debug("Delete called on %s filtered by (%s,%s). Target is \
  204. abstract, preparing reccursiv calls" % (target, filters, relational_filters))
  205. # Deletion with abstract LeObject as target (reccursiv calls)
  206. return self.__act_on_abstract(target, filters,
  207. relational_filters, self.delete)
  208. logger.debug("Delete called on %s filtered by (%s,%s)." % (
  209. target, filters, relational_filters))
  210. # Non abstract beahavior
  211. mongo_filters = self.__process_filters(
  212. target, filters, relational_filters)
  213. # Updating backref before deletion
  214. self.__update_backref_filtered(target, filters, relational_filters,
  215. None)
  216. res = self.__collection(target).remove(mongo_filters)
  217. return res['n']
  218. ## @brief updates records according to given filters
  219. # @param target Emclass : class of the object to insert
  220. # @param filters list : List of filters
  221. # @param relational_filters list : List of relational filters
  222. # @param upd_datas dict : datas to update (new values)
  223. # @return int : Number of updated records
  224. def update(self, target, filters, relational_filters, upd_datas):
  225. self._data_cast(upd_datas)
  226. #fetching current datas state
  227. mongo_filters = self.__process_filters(
  228. target, filters, relational_filters)
  229. old_datas_l = self.__collection(target).find(
  230. mongo_filters)
  231. old_datas_l = list(old_datas_l)
  232. #Running update
  233. res = self.__update_no_backref(target, filters, relational_filters,
  234. upd_datas)
  235. #updating backref
  236. self.__update_backref_filtered(target, filters, relational_filters,
  237. upd_datas, old_datas_l)
  238. return res
  239. ## @brief Designed to be called by backref update in order to avoid
  240. # infinite updates between back references
  241. # @see update()
  242. def __update_no_backref(self, target, filters, relational_filters,
  243. upd_datas):
  244. logger.debug("Update called on %s filtered by (%s,%s) with datas \
  245. %s" % (target, filters, relational_filters, upd_datas))
  246. if target.is_abstract():
  247. #Update using abstract LeObject as target (reccursiv calls)
  248. return self.__act_on_abstract(target, filters,
  249. relational_filters, self.update, upd_datas = upd_datas)
  250. #Non abstract beahavior
  251. mongo_filters = self.__process_filters(
  252. target, filters, relational_filters)
  253. self._data_cast(upd_datas)
  254. mongo_arg = {'$set': upd_datas }
  255. res = self.__collection(target).update(mongo_filters, mongo_arg)
  256. return res['n']
  257. ## @brief Inserts a record in a given collection
  258. # @param target Emclass : class of the object to insert
  259. # @param new_datas dict : datas to insert
  260. # @return the inserted uid
  261. def insert(self, target, new_datas):
  262. self._data_cast(new_datas)
  263. logger.debug("Insert called on %s with datas : %s"% (
  264. target, new_datas))
  265. uidname = target.uid_fieldname()[0] #MULTIPLE UID BROKEN HERE
  266. if uidname not in new_datas:
  267. raise MongoDataSourceError("Missing UID data will inserting a new \
  268. %s" % target.__class__)
  269. res = self.__collection(target).insert(new_datas)
  270. self.__update_backref(target, new_datas[uidname], None, new_datas)
  271. return str(res)
  272. ## @brief Inserts a list of records in a given collection
  273. # @param target Emclass : class of the objects inserted
  274. # @param datas_list list : list of dict
  275. # @return list : list of the inserted records' ids
  276. def insert_multi(self, target, datas_list):
  277. for datas in datas_list:
  278. self._data_cast(datas)
  279. res = self.__collection(target).insert_many(datas_list)
  280. for new_datas in datas_list:
  281. self.__update_backref(target, None, new_datas)
  282. target.make_consistency(datas=new_datas)
  283. return list(res.inserted_ids)
  284. ## @brief Update backref giving an action
  285. # @param target leObject child class
  286. # @param filters
  287. # @param relational_filters,
  288. # @param new_datas None | dict : optional new datas if None mean we are deleting
  289. # @param old_datas_l None | list : if None fetch old datas from db (usefull
  290. # when modifications are made on instance before updating backrefs)
  291. # @return nothing (for the moment
  292. def __update_backref_filtered(self, target,
  293. filters, relational_filters, new_datas = None, old_datas_l = None):
  294. # Getting all the UID of the object that will be deleted in order
  295. # to update back_references
  296. if old_datas_l is None:
  297. mongo_filters = self.__process_filters(
  298. target, filters, relational_filters)
  299. old_datas_l = self.__collection(target).find(
  300. mongo_filters)
  301. old_datas_l = list(old_datas_l)
  302. uidname = target.uid_fieldname()[0] # MULTIPLE UID BROKEN HERE
  303. for old_datas in old_datas_l:
  304. self.__update_backref(
  305. target, old_datas[uidname], old_datas, new_datas)
  306. ## @brief Update back references of an object
  307. # @ingroup plugin_mongodb_bref_op
  308. #
  309. # old_datas and new_datas arguments are set to None to indicate
  310. # insertion or deletion. Calls examples :
  311. # @par LeObject insert __update backref call
  312. # <pre>
  313. # Insert(datas):
  314. # self.make_insert(datas)
  315. # self.__update_backref(self.__class__, None, datas)
  316. # </pre>
  317. # @par LeObject delete __update backref call
  318. # Delete()
  319. # old_datas = self.datas()
  320. # self.make_delete()
  321. #  self.__update_backref(self.__class__, old_datas, None)
  322. # @par LeObject update __update_backref call
  323. # <pre>
  324. # Update(new_datas):
  325. # old_datas = self.datas()
  326. # self.make_udpdate(new_datas)
  327. #  self.__update_backref(self.__class__, old_datas, new_datas)
  328. # </pre>
  329. #
  330. # @param target LeObject child classa
  331. # @param tuid mixed : The target UID (the value that will be inserted in
  332. # back references)
  333. # @param old_datas dict : datas state before update
  334. # @param new_datas dict : datas state after the update process
  335. def __update_backref(self, target, tuid, old_datas, new_datas):
  336. #upd_dict is the dict that will allow to run updates in an optimized
  337. #way (or try to help doing it)
  338. #
  339. #Its structure looks like :
  340. # { LeoCLASS : {
  341. # UID1: (
  342. # LeoINSTANCE,
  343. # { fname1 : value, fname2: value }),
  344. # UID2 (LeoINSTANCE, {fname...}),
  345. # },
  346. # LeoClass2: {...
  347. #
  348. upd_dict = {}
  349. for fname, fdh in target.reference_handlers().items():
  350. oldd = old_datas is not None and fname in old_datas and \
  351. (not hasattr(fdh, 'default') or old_datas[fname] != fdh.default) \
  352. and not old_datas[fname] is None
  353. newd = new_datas is not None and fname in new_datas and \
  354. (not hasattr(fdh, 'default') or new_datas[fname] != fdh.default) \
  355. and not new_datas[fname] is None
  356. if (oldd and newd and old_datas[fname] == new_datas[fname])\
  357. or not(oldd or newd):
  358. # No changes or not concerned
  359. continue
  360. bref_cls = fdh.back_reference[0]
  361. bref_fname = fdh.back_reference[1]
  362. if not fdh.is_singlereference():
  363. # fdh is a multiple reference. So the update preparation will be
  364. # divided into two loops :
  365. # - one loop for deleting old datas
  366. # - one loop for inserting updated datas
  367. #
  368. # Preparing the list of values to delete or to add
  369. if newd and oldd:
  370. old_values = old_datas[fname]
  371. new_values = new_datas[fname]
  372. to_del = [ val
  373. for val in old_values
  374. if val not in new_values]
  375. to_add = [ val
  376. for val in new_values
  377. if val not in old_values]
  378. elif oldd and not newd:
  379. to_del = old_datas[fname]
  380. to_add = []
  381. elif not oldd and newd:
  382. to_del = []
  383. to_add = new_datas[fname]
  384. # Calling __back_ref_upd_one_value() with good arguments
  385. for vtype, vlist in [('old',to_del), ('new', to_add)]:
  386. for value in vlist:
  387. # fetching backref infos
  388. bref_infos = self.__bref_get_check(
  389. bref_cls, value, bref_fname)
  390. # preparing the upd_dict
  391. upd_dict = self.__update_backref_upd_dict_prepare(
  392. upd_dict, bref_infos, bref_fname, value)
  393. # preparing updated bref_infos
  394. bref_cls, bref_leo, bref_dh, bref_value = bref_infos
  395. bref_infos = (bref_cls, bref_leo, bref_dh,
  396. upd_dict[bref_cls][value][1][bref_fname])
  397. vdict = {vtype: value}
  398. # fetch and store updated value
  399. new_bref_val = self.__back_ref_upd_one_value(
  400. fname, fdh, tuid, bref_infos, **vdict)
  401. upd_dict[bref_cls][value][1][bref_fname] = new_bref_val
  402. else:
  403. # fdh is a single ref so the process is simpler, we do not have
  404. # to loop and we may do an update in only one
  405. # __back_ref_upd_one_value() call by giving both old and new
  406. # value
  407. vdict = {}
  408. if oldd:
  409. vdict['old'] = old_datas[fname]
  410. uid_val = vdict['old']
  411. if newd:
  412. vdict['new'] = new_datas[fname]
  413. if not oldd:
  414. uid_val = vdict['new']
  415. # Fetching back ref infos
  416. bref_infos = self.__bref_get_check(
  417. bref_cls, uid_val, bref_fname)
  418. # prepare the upd_dict
  419. upd_dict = self.__update_backref_upd_dict_prepare(
  420. upd_dict, bref_infos, bref_fname, uid_val)
  421. # forging update bref_infos
  422. bref_cls, bref_leo, bref_dh, bref_value = bref_infos
  423. bref_infos = (bref_cls, bref_leo, bref_dh,
  424. upd_dict[bref_cls][uid_val][1][bref_fname])
  425. # fetch and store updated value
  426. new_bref_val = self.__back_ref_upd_one_value(
  427. fname, fdh, tuid, bref_infos, **vdict)
  428. upd_dict[bref_cls][uid_val][1][bref_fname] = new_bref_val
  429. # Now we've got our upd_dict ready.
  430. # running the updates
  431. for bref_cls, uid_dict in upd_dict.items():
  432. for uidval, (leo, datas) in uid_dict.items():
  433. # MULTIPLE UID BROKEN 2 LINES BELOW
  434. self.__update_no_backref(
  435. leo.__class__, [(leo.uid_fieldname()[0], '=', uidval)],
  436. [], datas)
  437. ## @brief Utility function designed to handle the upd_dict of __update_backref()
  438. #
  439. # Basically checks if a key exists at some level, if not create it with
  440. # the good default value (in most case dict())
  441. # @param upd_dict dict : in & out args modified by reference
  442. # @param bref_infos tuple : as returned by __bref_get_check()
  443. # @param bref_fname str : name of the field in referenced class
  444. # @param uid_val mixed : the UID of the referenced object
  445. # @return the updated version of upd_dict
  446. @staticmethod
  447. def __update_backref_upd_dict_prepare(upd_dict,bref_infos, bref_fname,
  448. uid_val):
  449. bref_cls, bref_leo, bref_dh, bref_value = bref_infos
  450. if bref_cls not in upd_dict:
  451. upd_dict[bref_cls] = {}
  452. if uid_val not in upd_dict[bref_cls]:
  453. upd_dict[bref_cls][uid_val] = (bref_leo, {})
  454. if bref_fname not in upd_dict[bref_cls][uid_val]:
  455. upd_dict[bref_cls][uid_val][1][bref_fname] = bref_value
  456. return upd_dict
  457. ## @brief Prepare a one value back reference update
  458. # @param fname str : the source Reference field name
  459. # @param fdh DataHandler : the source Reference DataHandler
  460. # @param tuid mixed : the uid of the Leo that make reference to the backref
  461. # @param bref_infos tuple : as returned by __bref_get_check() method
  462. # @param old mixed : (optional **values) the old value
  463. # @param new mixed : (optional **values) the new value
  464. # @return the new back reference field value
  465. def __back_ref_upd_one_value(self, fname, fdh, tuid, bref_infos, **values):
  466. bref_cls, bref_leo, bref_dh, bref_val = bref_infos
  467. oldd = 'old' in values
  468. newdd = 'new' in values
  469. if bref_val is None:
  470. bref_val = bref_dh.empty()
  471. if not bref_dh.is_singlereference():
  472. if oldd and newdd:
  473. if tuid not in bref_val:
  474. raise MongoDbConsistencyError("The value we want to \
  475. delete in this back reference update was not found in the back referenced \
  476. object : %s. Value was : '%s'" % (bref_leo, tuid))
  477. return bref_val
  478. elif oldd and not newdd:
  479. # deletion
  480. old_value = values['old']
  481. if tuid not in bref_val:
  482. raise MongoDbConsistencyError("The value we want to \
  483. delete in this back reference update was not found in the back referenced \
  484. object : %s. Value was : '%s'" % (bref_leo, tuid))
  485. if isinstance(bref_val, tuple):
  486. bref_val = set(bref_val)
  487. if isinstance(bref_val, set):
  488. bref_val -= set([tuid])
  489. else:
  490. del(bref_val[bref_val.index(tuid)])
  491. elif not oldd and newdd:
  492. if tuid in bref_val:
  493. raise MongoDbConsistencyError("The value we want to \
  494. add in this back reference update was found in the back referenced \
  495. object : %s. Value was : '%s'" % (bref_leo, tuid))
  496. if isinstance(bref_val, tuple):
  497. bref_val = set(bref_val)
  498. if isinstance(bref_val, set):
  499. bref_val |= set([tuid])
  500. else:
  501. bref_val.append(tuid)
  502. else:
  503. # Single value backref
  504. if oldd and newdd:
  505. if bref_val != tuid:
  506. raise MongoDbConsistencyError("The backreference doesn't \
  507. have expected value. Expected was %s but found %s in %s" % (
  508. tuid, bref_val, bref_leo))
  509. return bref_val
  510. elif oldd and not newdd:
  511. # deletion
  512. if not hasattr(bref_dh, "default"):
  513. raise MongoDbConsistencyError("Unable to delete a \
  514. value for a back reference update. The concerned field don't have a default \
  515. value : in %s field %s" % (bref_leo,fname))
  516. bref_val = getattr(bref_dh, "default")
  517. elif not oldd and newdd:
  518. bref_val = tuid
  519. return bref_val
  520. ## @brief Fetch back reference informations
  521. # @warning thank's to __update_backref_act() this method is useless
  522. # @param bref_cls LeObject child class : __back_reference[0]
  523. # @param uidv mixed : UID value (the content of the reference field)
  524. # @param bref_fname str : the name of the back_reference field
  525. # @return tuple(bref_class, bref_LeObect_instance, bref_datahandler, bref_value)
  526. # @throw MongoDbConsistencyError when LeObject instance not found given uidv
  527. # @throw LodelFatalError if the back reference field is not a Reference subclass (major failure)
  528. def __bref_get_check(self, bref_cls, uidv, bref_fname):
  529. bref_leo = bref_cls.get_from_uid(uidv)
  530. if bref_leo is None:
  531. raise MongoDbConsistencyError("Unable to get the object we make \
  532. reference to : %s with uid = %s" % (bref_cls, repr(uidv)))
  533. bref_dh = bref_leo.data_handler(bref_fname)
  534. if not bref_dh.is_reference():
  535. raise LodelFatalError("Found a back reference field that \
  536. is not a reference : '%s' field '%s'" % (bref_leo, bref_fname))
  537. bref_val = bref_leo.data(bref_fname)
  538. return (bref_leo.__class__, bref_leo, bref_dh, bref_val)
  539. ## @brief Act on abstract LeObject child
  540. #
  541. # This method is designed to be called by insert, select and delete method
  542. # when they encounter an abtract class
  543. # @param target LeObject child class
  544. # @param filters
  545. # @param relational_filters
  546. # @param act function : the caller method
  547. # @param **kwargs other arguments
  548. # @return sum of results (if it's an array it will result in a concat)
  549. # @todo optimization implementing a cache for __bref_get_check()
  550. def __act_on_abstract(self,
  551. target, filters, relational_filters, act, **kwargs):
  552. logger.debug("Abstract %s, running reccursiv select \
  553. on non abstract childs" % act.__name__)
  554. result = list() if act == self.select else 0
  555. if not target.is_abstract():
  556. target_childs = [target]
  557. else:
  558. target_childs = [tc for tc in target.child_classes()
  559. if not tc.is_abstract()]
  560. for target_child in target_childs:
  561. logger.debug(
  562. "Abstract %s on %s" % (act.__name__, target_child.__name__))
  563. # Add target_child to filter
  564. new_filters = copy.copy(filters)
  565. for i in range(len(filters)):
  566. fname, op, val = filters[i]
  567. if fname == CLASS_ID_FIELDNAME:
  568. logger.warning("Dirty drop of filter : '%s %s %s'" % (
  569. fname, op, val))
  570. del(new_filters[i])
  571. new_filters.append(
  572. (CLASS_ID_FIELDNAME, '=',
  573. collection_name(target_child.__name__)))
  574. result += act(
  575. target = target_child,
  576. filters = new_filters,
  577. relational_filters = relational_filters,
  578. **kwargs)
  579. return result
  580. ## @brief Connect to database
  581. # @note this method avoid opening two times the same connection using MongoDbDatasource::_connections static attribute
  582. # @param username str
  583. # @param password str
  584. # @param ro bool : If True the Datasource is for read only, else it will be write only
  585. def __connect(self, username, password, db_name, ro):
  586. conn_string = connection_string(
  587. username = username, password = password,
  588. host = self.__db_infos['host'],
  589. port = self.__db_infos['port'],
  590. db_name = db_name,
  591. ro = ro)
  592. self.__conn_hash = conn_h = hash(conn_string)
  593. if conn_h in self._connections:
  594. self._connections[conn_h]['conn_count'] += 1
  595. return self._connections[conn_h]['db'][self.__db_infos['db_name']]
  596. else:
  597. logger.info("Opening a new connection to database")
  598. self._connections[conn_h] = {
  599. 'conn_count': 1,
  600. 'db': utils.connect(conn_string)}
  601. return self._connections[conn_h]['db'][self.__db_infos['db_name']]
  602. ## @brief Return a pymongo collection given a LeObject child class
  603. # @param leobject LeObject child class (no instance)
  604. # @return a pymongo.collection instance
  605. def __collection(self, leobject):
  606. return self.database[object_collection_name(leobject)]
  607. ## @brief Perform subqueries implies by relational filters and append the
  608. # result to existing filters
  609. #
  610. # The processing is divided in multiple steps :
  611. # - determine (for each relational field of the target) every collection that are involved
  612. # - generate subqueries for relational_filters that concerns a different collection than target collection filters
  613. # - execute subqueries
  614. # - transform subqueries results in filters
  615. # - merge subqueries generated filters with existing filters
  616. #
  617. # @param target LeObject subclass (no instance) : Target class
  618. # @param filters list : List of tuple(FIELDNAME, OP, VALUE)
  619. # @param relational_filters : same composition thant filters except that FIELD is represented by a tuple(FIELDNAME, {CLASS1:RFIELD1, CLASS2:RFIELD2})
  620. # @return a list of pymongo filters ( dict {FIELD:{OPERATOR:VALUE}} )
  621. def __process_filters(self,target, filters, relational_filters):
  622. # Simple filters lodel2 -> pymongo converting
  623. res = self.__filters2mongo(filters, target)
  624. rfilters = self.__prepare_relational_filters(target, relational_filters)
  625. #Now that everything is well organized, begin to forge subquerie
  626. #filters
  627. self.__subqueries_from_relational_filters(target, rfilters)
  628. # Executing subqueries, creating filters from result, and injecting
  629. # them in original filters of the query
  630. if len(rfilters) > 0:
  631. logger.debug("Begining subquery execution")
  632. for fname in rfilters:
  633. if fname not in res:
  634. res[fname] = dict()
  635. subq_results = set()
  636. for leobject, sq_filters in rfilters[fname].items():
  637. uid_fname = mongo_fieldname(leobject._uid)
  638. log_msg = "Subquery running on collection {coll} with filters \
  639. '{filters}'"
  640. logger.debug(log_msg.format(
  641. coll=object_collection_name(leobject),
  642. filters=sq_filters))
  643. cursor = self.__collection(leobject).find(
  644. filter=sq_filters,
  645. projection=uid_fname)
  646. subq_results |= set(doc[uid_fname] for doc in cursor)
  647. #generating new filter from result
  648. if '$in' in res[fname]:
  649. #WARNING we allready have a IN on this field, doing dedup
  650. #from result
  651. deduped = set(res[fname]['$in']) & subq_results
  652. if len(deduped) == 0:
  653. del(res[fname]['$in'])
  654. else:
  655. res[fname]['$in'] = list(deduped)
  656. else:
  657. res[fname]['$in'] = list(subq_results)
  658. if len(rfilters) > 0:
  659. logger.debug("End of subquery execution")
  660. return res
  661. ## @brief Generate subqueries from rfilters tree
  662. #
  663. # Returned struct organization :
  664. # - 1st level keys : relational field name of target
  665. # - 2nd level keys : referenced leobject
  666. # - 3th level values : pymongo filters (dict)
  667. #
  668. # @note The only caller of this method is __process_filters
  669. # @warning No return value, the rfilters arguement is modified by reference
  670. #
  671. # @param target LeObject subclass (no instance) : Target class
  672. # @param rfilters dict : A struct as returned by MongoDbDatasource.__prepare_relational_filters()
  673. # @return None, the rfilters argument is modified by reference
  674. @classmethod
  675. def __subqueries_from_relational_filters(cls, target, rfilters):
  676. for fname in rfilters:
  677. for leobject in rfilters[fname]:
  678. for rfield in rfilters[fname][leobject]:
  679. #This way of doing is not optimized but allows to trigger
  680. #warnings in some case (2 different values for a same op
  681. #on a same field on a same collection)
  682. mongofilters = cls.__op_value_listconv(
  683. rfilters[fname][leobject][rfield], target.field(fname))
  684. rfilters[fname][leobject][rfield] = mongofilters
  685. ## @brief Generate a tree from relational_filters
  686. #
  687. # The generated struct is a dict with :
  688. # - 1st level keys : relational field name of target
  689. # - 2nd level keys : referenced leobject
  690. # - 3th level keys : referenced field in referenced class
  691. # - 4th level values : list of tuple(op, value)
  692. #
  693. # @note The only caller of this method is __process_filters
  694. # @warning An assertion is done : if two leobject are stored in the same collection they share the same uid
  695. #
  696. # @param target LeObject subclass (no instance) : Target class
  697. # @param relational_filters : same composition thant filters except that
  698. # @return a struct as described above
  699. @classmethod
  700. def __prepare_relational_filters(cls, target, relational_filters):
  701. # We are going to regroup relationnal filters by reference field
  702. # then by collection
  703. rfilters = dict()
  704. if relational_filters is None:
  705. relational_filters = []
  706. for (fname, rfields), op, value in relational_filters:
  707. if fname not in rfilters:
  708. rfilters[fname] = dict()
  709. rfilters[fname] = dict()
  710. # Stores the representative leobject for associated to a collection
  711. # name
  712. leo_collname = dict()
  713. # WARNING ! Here we assert that all leobject that are stored
  714. # in a same collection are identified by the same field
  715. for leobject, rfield in rfields.items():
  716. #here we are filling a dict with leobject as index but
  717. #we are doing a UNIQ on collection name
  718. cur_collname = object_collection_name(leobject)
  719. if cur_collname not in leo_collname:
  720. leo_collname[cur_collname] = leobject
  721. rfilters[fname][leobject] = dict()
  722. #Fecthing the collection's representative leobject
  723. repr_leo = leo_collname[cur_collname]
  724. if rfield not in rfilters[fname][repr_leo]:
  725. rfilters[fname][repr_leo][rfield] = list()
  726. rfilters[fname][repr_leo][rfield].append((op, value))
  727. return rfilters
  728. ## @brief Convert lodel2 filters to pymongo conditions
  729. # @param filters list : list of lodel filters
  730. # @return dict representing pymongo conditions
  731. @classmethod
  732. def __filters2mongo(cls, filters, target):
  733. res = dict()
  734. eq_fieldname = [] #Stores field with equal comparison OP
  735. for fieldname, op, value in filters:
  736. oop = op
  737. ovalue = value
  738. op, value = cls.__op_value_conv(op, value, target.field(fieldname))
  739. if op == '=':
  740. eq_fieldname.append(fieldname)
  741. if fieldname in res:
  742. logger.warning("Dropping previous condition. Overwritten \
  743. by an equality filter")
  744. res[fieldname] = value
  745. continue
  746. if fieldname in eq_fieldname:
  747. logger.warning("Dropping condition : '%s %s %s'" % (
  748. fieldname, op, value))
  749. continue
  750. if fieldname not in res:
  751. res[fieldname] = dict()
  752. if op in res[fieldname]:
  753. logger.warning("Dropping condition : '%s %s %s'" % (
  754. fieldname, op, value))
  755. else:
  756. if op not in cls.lodel2mongo_op_map:
  757. raise ValueError("Invalid operator : '%s'" % op)
  758. new_op = cls.lodel2mongo_op_map[op]
  759. res[fieldname][new_op] = value
  760. return res
  761. ## @brief Convert lodel2 operator and value to pymongo struct
  762. #
  763. # Convertion is done using MongoDbDatasource::lodel2mongo_op_map
  764. # @param op str : take value in LeFilteredQuery::_query_operators
  765. # @param value mixed : the value
  766. # @param dhdl
  767. # @return a tuple(mongo_op, mongo_value)
  768. @classmethod
  769. def __op_value_conv(cls, op, value, dhdl):
  770. if op not in cls.lodel2mongo_op_map:
  771. msg = "Invalid operator '%s' found" % op
  772. raise MongoDbDataSourceError(msg)
  773. mongop = cls.lodel2mongo_op_map[op]
  774. mongoval = value
  775. # Converting lodel2 wildcarded string into a case insensitive
  776. # mongodb re
  777. if mongop in cls.mongo_op_re:
  778. if value.startswith('(') and value.endswith(')'):
  779. if (dhdl.cast_type is not None):
  780. mongoval = [ dhdl.cast_type(item) for item in mongoval[1:-1].split(',') ]
  781. else:
  782. mongoval = [ item for item in mongoval[1:-1].split(',') ]
  783. elif mongop == 'like':
  784. #unescaping \
  785. mongoval = value.replace('\\\\','\\')
  786. if not mongoval.startswith('*'):
  787. mongoval = '^'+mongoval
  788. #For the end of the string it's harder to detect escaped *
  789. if not (mongoval[-1] == '*' and mongoval[-2] != '\\'):
  790. mongoval += '$'
  791. #Replacing every other unescaped wildcard char
  792. mongoval = cls.wildcard_re.sub('.*', mongoval)
  793. mongoval = {'$regex': mongoval, '$options': 'i'}
  794. return (op, mongoval)
  795. ## @brief Convert a list of tuple(OP, VALUE) into a pymongo filter dict
  796. # @param op_value_list list
  797. # @param dhdl
  798. # @return a dict with mongo op as key and value as value...
  799. @classmethod
  800. def __op_value_listconv(cls, op_value_list, dhdl):
  801. result = dict()
  802. for op, value in op_value_list:
  803. mongop, mongoval = cls.__op_value_conv(op, value, dhdl)
  804. if mongop in result:
  805. warnings.warn("Duplicated value given for a single \
  806. field/operator couple in a query. We will keep only the first one")
  807. else:
  808. result[mongop] = mongoval
  809. return result
  810. ##@brief Generate a comparison function for post reccursion sorting in
  811. #select
  812. #@return a lambda function that take 2 dict as arguement
  813. @classmethod
  814. def __generate_lambda_cmp_order(cls, order):
  815. if len(order) == 0:
  816. return lambda a,b: 0
  817. glco = cls.__generate_lambda_cmp_order
  818. fname, cmpdir = order[0]
  819. order = order[1:]
  820. return lambda a,b: glco(order)(a,b) if a[fname] == b[fname] else (\
  821. 1 if (a[fname]>b[fname] if cmpdir == 'ASC' else a[fname]<b[fname])\
  822. else -1)
  823. ##@brief Correct some datas before giving them to pymongo
  824. #
  825. #For example sets has to be casted to lise
  826. #@param datas
  827. #@return datas
  828. @classmethod
  829. def _data_cast(cls, datas):
  830. for dname in datas:
  831. if isinstance(datas[dname], set):
  832. #pymongo raises :
  833. #bson.errors.InvalidDocument: Cannot encode object: {...}
  834. #with sets
  835. datas[dname] = list(datas[dname])
  836. return datas