cursor.py 37 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044
  1. # Copyright 2009-2015 MongoDB, Inc.
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. """Cursor class to iterate over Mongo query results."""
  15. import copy
  16. from collections import deque
  17. from bson import RE_TYPE
  18. from bson.code import Code
  19. from bson.py3compat import (iteritems,
  20. integer_types,
  21. string_type)
  22. from bson.son import SON
  23. from pymongo import helpers
  24. from pymongo.common import validate_boolean, validate_is_mapping
  25. from pymongo.errors import (AutoReconnect,
  26. ConnectionFailure,
  27. InvalidOperation,
  28. NotMasterError,
  29. OperationFailure)
  30. from pymongo.message import _GetMore, _Query
  31. from pymongo.read_preferences import ReadPreference
  32. _QUERY_OPTIONS = {
  33. "tailable_cursor": 2,
  34. "slave_okay": 4,
  35. "oplog_replay": 8,
  36. "no_timeout": 16,
  37. "await_data": 32,
  38. "exhaust": 64,
  39. "partial": 128}
  40. class CursorType(object):
  41. NON_TAILABLE = 0
  42. """The standard cursor type."""
  43. TAILABLE = _QUERY_OPTIONS["tailable_cursor"]
  44. """The tailable cursor type.
  45. Tailable cursors are only for use with capped collections. They are not
  46. closed when the last data is retrieved but are kept open and the cursor
  47. location marks the final document position. If more data is received
  48. iteration of the cursor will continue from the last document received.
  49. """
  50. TAILABLE_AWAIT = TAILABLE | _QUERY_OPTIONS["await_data"]
  51. """A tailable cursor with the await option set.
  52. Creates a tailable cursor that will wait for a few seconds after returning
  53. the full result set so that it can capture and return additional data added
  54. during the query.
  55. """
  56. EXHAUST = _QUERY_OPTIONS["exhaust"]
  57. """An exhaust cursor.
  58. MongoDB will stream batched results to the client without waiting for the
  59. client to request each batch, reducing latency.
  60. """
  61. # This has to be an old style class due to
  62. # http://bugs.jython.org/issue1057
  63. class _SocketManager:
  64. """Used with exhaust cursors to ensure the socket is returned.
  65. """
  66. def __init__(self, sock, pool):
  67. self.sock = sock
  68. self.pool = pool
  69. self.__closed = False
  70. def __del__(self):
  71. self.close()
  72. def close(self):
  73. """Return this instance's socket to the connection pool.
  74. """
  75. if not self.__closed:
  76. self.__closed = True
  77. self.pool.return_socket(self.sock)
  78. self.sock, self.pool = None, None
  79. class Cursor(object):
  80. """A cursor / iterator over Mongo query results.
  81. """
  82. def __init__(self, collection, filter=None, projection=None, skip=0,
  83. limit=0, no_cursor_timeout=False,
  84. cursor_type=CursorType.NON_TAILABLE,
  85. sort=None, allow_partial_results=False, oplog_replay=False,
  86. modifiers=None, batch_size=0, manipulate=True):
  87. """Create a new cursor.
  88. Should not be called directly by application developers - see
  89. :meth:`~pymongo.collection.Collection.find` instead.
  90. .. mongodoc:: cursors
  91. """
  92. self.__id = None
  93. spec = filter
  94. if spec is None:
  95. spec = {}
  96. validate_is_mapping("filter", spec)
  97. if not isinstance(skip, int):
  98. raise TypeError("skip must be an instance of int")
  99. if not isinstance(limit, int):
  100. raise TypeError("limit must be an instance of int")
  101. validate_boolean("no_cursor_timeout", no_cursor_timeout)
  102. if cursor_type not in (CursorType.NON_TAILABLE, CursorType.TAILABLE,
  103. CursorType.TAILABLE_AWAIT, CursorType.EXHAUST):
  104. raise ValueError("not a valid value for cursor_type")
  105. validate_boolean("allow_partial_results", allow_partial_results)
  106. validate_boolean("oplog_replay", oplog_replay)
  107. if modifiers is not None:
  108. validate_is_mapping("modifiers", modifiers)
  109. if not isinstance(batch_size, integer_types):
  110. raise TypeError("batch_size must be an integer")
  111. if batch_size < 0:
  112. raise ValueError("batch_size must be >= 0")
  113. if projection is not None:
  114. if not projection:
  115. projection = {"_id": 1}
  116. projection = helpers._fields_list_to_dict(projection, "projection")
  117. self.__collection = collection
  118. self.__spec = spec
  119. self.__projection = projection
  120. self.__skip = skip
  121. self.__limit = limit
  122. self.__batch_size = batch_size
  123. self.__modifiers = modifiers and modifiers.copy() or {}
  124. self.__ordering = sort and helpers._index_document(sort) or None
  125. self.__max_scan = None
  126. self.__explain = False
  127. self.__hint = None
  128. self.__comment = None
  129. self.__max_time_ms = None
  130. self.__max = None
  131. self.__min = None
  132. self.__manipulate = manipulate
  133. # Exhaust cursor support
  134. self.__exhaust = False
  135. self.__exhaust_mgr = None
  136. if cursor_type == CursorType.EXHAUST:
  137. if self.__collection.database.client.is_mongos:
  138. raise InvalidOperation('Exhaust cursors are '
  139. 'not supported by mongos')
  140. if limit:
  141. raise InvalidOperation("Can't use limit and exhaust together.")
  142. self.__exhaust = True
  143. # This is ugly. People want to be able to do cursor[5:5] and
  144. # get an empty result set (old behavior was an
  145. # exception). It's hard to do that right, though, because the
  146. # server uses limit(0) to mean 'no limit'. So we set __empty
  147. # in that case and check for it when iterating. We also unset
  148. # it anytime we change __limit.
  149. self.__empty = False
  150. self.__data = deque()
  151. self.__address = None
  152. self.__retrieved = 0
  153. self.__killed = False
  154. self.__codec_options = collection.codec_options
  155. self.__read_preference = collection.read_preference
  156. self.__query_flags = cursor_type
  157. if self.__read_preference != ReadPreference.PRIMARY:
  158. self.__query_flags |= _QUERY_OPTIONS["slave_okay"]
  159. if no_cursor_timeout:
  160. self.__query_flags |= _QUERY_OPTIONS["no_timeout"]
  161. if allow_partial_results:
  162. self.__query_flags |= _QUERY_OPTIONS["partial"]
  163. if oplog_replay:
  164. self.__query_flags |= _QUERY_OPTIONS["oplog_replay"]
  165. @property
  166. def collection(self):
  167. """The :class:`~pymongo.collection.Collection` that this
  168. :class:`Cursor` is iterating.
  169. """
  170. return self.__collection
  171. @property
  172. def retrieved(self):
  173. """The number of documents retrieved so far.
  174. """
  175. return self.__retrieved
  176. def __del__(self):
  177. if self.__id and not self.__killed:
  178. self.__die()
  179. def rewind(self):
  180. """Rewind this cursor to its unevaluated state.
  181. Reset this cursor if it has been partially or completely evaluated.
  182. Any options that are present on the cursor will remain in effect.
  183. Future iterating performed on this cursor will cause new queries to
  184. be sent to the server, even if the resultant data has already been
  185. retrieved by this cursor.
  186. """
  187. self.__data = deque()
  188. self.__id = None
  189. self.__address = None
  190. self.__retrieved = 0
  191. self.__killed = False
  192. return self
  193. def clone(self):
  194. """Get a clone of this cursor.
  195. Returns a new Cursor instance with options matching those that have
  196. been set on the current instance. The clone will be completely
  197. unevaluated, even if the current instance has been partially or
  198. completely evaluated.
  199. """
  200. return self._clone(True)
  201. def _clone(self, deepcopy=True):
  202. """Internal clone helper."""
  203. clone = self._clone_base()
  204. values_to_clone = ("spec", "projection", "skip", "limit",
  205. "max_time_ms", "comment", "max", "min",
  206. "ordering", "explain", "hint", "batch_size",
  207. "max_scan", "manipulate", "query_flags",
  208. "modifiers")
  209. data = dict((k, v) for k, v in iteritems(self.__dict__)
  210. if k.startswith('_Cursor__') and k[9:] in values_to_clone)
  211. if deepcopy:
  212. data = self._deepcopy(data)
  213. clone.__dict__.update(data)
  214. return clone
  215. def _clone_base(self):
  216. """Creates an empty Cursor object for information to be copied into.
  217. """
  218. return Cursor(self.__collection)
  219. def __die(self):
  220. """Closes this cursor.
  221. """
  222. if self.__id and not self.__killed:
  223. if self.__exhaust and self.__exhaust_mgr:
  224. # If this is an exhaust cursor and we haven't completely
  225. # exhausted the result set we *must* close the socket
  226. # to stop the server from sending more data.
  227. self.__exhaust_mgr.sock.close()
  228. else:
  229. self.__collection.database.client.close_cursor(self.__id,
  230. self.__address)
  231. if self.__exhaust and self.__exhaust_mgr:
  232. self.__exhaust_mgr.close()
  233. self.__killed = True
  234. def close(self):
  235. """Explicitly close / kill this cursor. Required for PyPy, Jython and
  236. other Python implementations that don't use reference counting
  237. garbage collection.
  238. """
  239. self.__die()
  240. def __query_spec(self):
  241. """Get the spec to use for a query.
  242. """
  243. operators = self.__modifiers.copy()
  244. if self.__ordering:
  245. operators["$orderby"] = self.__ordering
  246. if self.__explain:
  247. operators["$explain"] = True
  248. if self.__hint:
  249. operators["$hint"] = self.__hint
  250. if self.__comment:
  251. operators["$comment"] = self.__comment
  252. if self.__max_scan:
  253. operators["$maxScan"] = self.__max_scan
  254. if self.__max_time_ms is not None:
  255. operators["$maxTimeMS"] = self.__max_time_ms
  256. if self.__max:
  257. operators["$max"] = self.__max
  258. if self.__min:
  259. operators["$min"] = self.__min
  260. if operators:
  261. # Make a shallow copy so we can cleanly rewind or clone.
  262. spec = self.__spec.copy()
  263. # White-listed commands must be wrapped in $query.
  264. if "$query" not in spec:
  265. # $query has to come first
  266. spec = SON([("$query", spec)])
  267. if not isinstance(spec, SON):
  268. # Ensure the spec is SON. As order is important this will
  269. # ensure its set before merging in any extra operators.
  270. spec = SON(spec)
  271. spec.update(operators)
  272. return spec
  273. # Have to wrap with $query if "query" is the first key.
  274. # We can't just use $query anytime "query" is a key as
  275. # that breaks commands like count and find_and_modify.
  276. # Checking spec.keys()[0] covers the case that the spec
  277. # was passed as an instance of SON or OrderedDict.
  278. elif ("query" in self.__spec and
  279. (len(self.__spec) == 1 or
  280. next(iter(self.__spec)) == "query")):
  281. return SON({"$query": self.__spec})
  282. return self.__spec
  283. def __check_okay_to_chain(self):
  284. """Check if it is okay to chain more options onto this cursor.
  285. """
  286. if self.__retrieved or self.__id is not None:
  287. raise InvalidOperation("cannot set options after executing query")
  288. def add_option(self, mask):
  289. """Set arbitrary query flags using a bitmask.
  290. To set the tailable flag:
  291. cursor.add_option(2)
  292. """
  293. if not isinstance(mask, int):
  294. raise TypeError("mask must be an int")
  295. self.__check_okay_to_chain()
  296. if mask & _QUERY_OPTIONS["exhaust"]:
  297. if self.__limit:
  298. raise InvalidOperation("Can't use limit and exhaust together.")
  299. if self.__collection.database.client.is_mongos:
  300. raise InvalidOperation('Exhaust cursors are '
  301. 'not supported by mongos')
  302. self.__exhaust = True
  303. self.__query_flags |= mask
  304. return self
  305. def remove_option(self, mask):
  306. """Unset arbitrary query flags using a bitmask.
  307. To unset the tailable flag:
  308. cursor.remove_option(2)
  309. """
  310. if not isinstance(mask, int):
  311. raise TypeError("mask must be an int")
  312. self.__check_okay_to_chain()
  313. if mask & _QUERY_OPTIONS["exhaust"]:
  314. self.__exhaust = False
  315. self.__query_flags &= ~mask
  316. return self
  317. def limit(self, limit):
  318. """Limits the number of results to be returned by this cursor.
  319. Raises :exc:`TypeError` if `limit` is not an integer. Raises
  320. :exc:`~pymongo.errors.InvalidOperation` if this :class:`Cursor`
  321. has already been used. The last `limit` applied to this cursor
  322. takes precedence. A limit of ``0`` is equivalent to no limit.
  323. :Parameters:
  324. - `limit`: the number of results to return
  325. .. mongodoc:: limit
  326. """
  327. if not isinstance(limit, integer_types):
  328. raise TypeError("limit must be an integer")
  329. if self.__exhaust:
  330. raise InvalidOperation("Can't use limit and exhaust together.")
  331. self.__check_okay_to_chain()
  332. self.__empty = False
  333. self.__limit = limit
  334. return self
  335. def batch_size(self, batch_size):
  336. """Limits the number of documents returned in one batch. Each batch
  337. requires a round trip to the server. It can be adjusted to optimize
  338. performance and limit data transfer.
  339. .. note:: batch_size can not override MongoDB's internal limits on the
  340. amount of data it will return to the client in a single batch (i.e
  341. if you set batch size to 1,000,000,000, MongoDB will currently only
  342. return 4-16MB of results per batch).
  343. Raises :exc:`TypeError` if `batch_size` is not an integer.
  344. Raises :exc:`ValueError` if `batch_size` is less than ``0``.
  345. Raises :exc:`~pymongo.errors.InvalidOperation` if this
  346. :class:`Cursor` has already been used. The last `batch_size`
  347. applied to this cursor takes precedence.
  348. :Parameters:
  349. - `batch_size`: The size of each batch of results requested.
  350. """
  351. if not isinstance(batch_size, integer_types):
  352. raise TypeError("batch_size must be an integer")
  353. if batch_size < 0:
  354. raise ValueError("batch_size must be >= 0")
  355. self.__check_okay_to_chain()
  356. self.__batch_size = batch_size == 1 and 2 or batch_size
  357. return self
  358. def skip(self, skip):
  359. """Skips the first `skip` results of this cursor.
  360. Raises :exc:`TypeError` if `skip` is not an integer. Raises
  361. :exc:`ValueError` if `skip` is less than ``0``. Raises
  362. :exc:`~pymongo.errors.InvalidOperation` if this :class:`Cursor` has
  363. already been used. The last `skip` applied to this cursor takes
  364. precedence.
  365. :Parameters:
  366. - `skip`: the number of results to skip
  367. """
  368. if not isinstance(skip, integer_types):
  369. raise TypeError("skip must be an integer")
  370. if skip < 0:
  371. raise ValueError("skip must be >= 0")
  372. self.__check_okay_to_chain()
  373. self.__skip = skip
  374. return self
  375. def max_time_ms(self, max_time_ms):
  376. """Specifies a time limit for a query operation. If the specified
  377. time is exceeded, the operation will be aborted and
  378. :exc:`~pymongo.errors.ExecutionTimeout` is raised. If `max_time_ms`
  379. is ``None`` no limit is applied.
  380. Raises :exc:`TypeError` if `max_time_ms` is not an integer or ``None``.
  381. Raises :exc:`~pymongo.errors.InvalidOperation` if this :class:`Cursor`
  382. has already been used.
  383. :Parameters:
  384. - `max_time_ms`: the time limit after which the operation is aborted
  385. """
  386. if (not isinstance(max_time_ms, integer_types)
  387. and max_time_ms is not None):
  388. raise TypeError("max_time_ms must be an integer or None")
  389. self.__check_okay_to_chain()
  390. self.__max_time_ms = max_time_ms
  391. return self
  392. def __getitem__(self, index):
  393. """Get a single document or a slice of documents from this cursor.
  394. Raises :class:`~pymongo.errors.InvalidOperation` if this
  395. cursor has already been used.
  396. To get a single document use an integral index, e.g.::
  397. >>> db.test.find()[50]
  398. An :class:`IndexError` will be raised if the index is negative
  399. or greater than the amount of documents in this cursor. Any
  400. limit previously applied to this cursor will be ignored.
  401. To get a slice of documents use a slice index, e.g.::
  402. >>> db.test.find()[20:25]
  403. This will return this cursor with a limit of ``5`` and skip of
  404. ``20`` applied. Using a slice index will override any prior
  405. limits or skips applied to this cursor (including those
  406. applied through previous calls to this method). Raises
  407. :class:`IndexError` when the slice has a step, a negative
  408. start value, or a stop value less than or equal to the start
  409. value.
  410. :Parameters:
  411. - `index`: An integer or slice index to be applied to this cursor
  412. """
  413. self.__check_okay_to_chain()
  414. self.__empty = False
  415. if isinstance(index, slice):
  416. if index.step is not None:
  417. raise IndexError("Cursor instances do not support slice steps")
  418. skip = 0
  419. if index.start is not None:
  420. if index.start < 0:
  421. raise IndexError("Cursor instances do not support"
  422. "negative indices")
  423. skip = index.start
  424. if index.stop is not None:
  425. limit = index.stop - skip
  426. if limit < 0:
  427. raise IndexError("stop index must be greater than start"
  428. "index for slice %r" % index)
  429. if limit == 0:
  430. self.__empty = True
  431. else:
  432. limit = 0
  433. self.__skip = skip
  434. self.__limit = limit
  435. return self
  436. if isinstance(index, integer_types):
  437. if index < 0:
  438. raise IndexError("Cursor instances do not support negative"
  439. "indices")
  440. clone = self.clone()
  441. clone.skip(index + self.__skip)
  442. clone.limit(-1) # use a hard limit
  443. for doc in clone:
  444. return doc
  445. raise IndexError("no such item for Cursor instance")
  446. raise TypeError("index %r cannot be applied to Cursor "
  447. "instances" % index)
  448. def max_scan(self, max_scan):
  449. """Limit the number of documents to scan when performing the query.
  450. Raises :class:`~pymongo.errors.InvalidOperation` if this
  451. cursor has already been used. Only the last :meth:`max_scan`
  452. applied to this cursor has any effect.
  453. :Parameters:
  454. - `max_scan`: the maximum number of documents to scan
  455. """
  456. self.__check_okay_to_chain()
  457. self.__max_scan = max_scan
  458. return self
  459. def max(self, spec):
  460. """Adds `max` operator that specifies upper bound for specific index.
  461. :Parameters:
  462. - `spec`: a list of field, limit pairs specifying the exclusive
  463. upper bound for all keys of a specific index in order.
  464. .. versionadded:: 2.7
  465. """
  466. if not isinstance(spec, (list, tuple)):
  467. raise TypeError("spec must be an instance of list or tuple")
  468. self.__check_okay_to_chain()
  469. self.__max = SON(spec)
  470. return self
  471. def min(self, spec):
  472. """Adds `min` operator that specifies lower bound for specific index.
  473. :Parameters:
  474. - `spec`: a list of field, limit pairs specifying the inclusive
  475. lower bound for all keys of a specific index in order.
  476. .. versionadded:: 2.7
  477. """
  478. if not isinstance(spec, (list, tuple)):
  479. raise TypeError("spec must be an instance of list or tuple")
  480. self.__check_okay_to_chain()
  481. self.__min = SON(spec)
  482. return self
  483. def sort(self, key_or_list, direction=None):
  484. """Sorts this cursor's results.
  485. Pass a field name and a direction, either
  486. :data:`~pymongo.ASCENDING` or :data:`~pymongo.DESCENDING`::
  487. for doc in collection.find().sort('field', pymongo.ASCENDING):
  488. print(doc)
  489. To sort by multiple fields, pass a list of (key, direction) pairs::
  490. for doc in collection.find().sort([
  491. ('field1', pymongo.ASCENDING),
  492. ('field2', pymongo.DESCENDING)]):
  493. print(doc)
  494. Beginning with MongoDB version 2.6, text search results can be
  495. sorted by relevance::
  496. cursor = db.test.find(
  497. {'$text': {'$search': 'some words'}},
  498. {'score': {'$meta': 'textScore'}})
  499. # Sort by 'score' field.
  500. cursor.sort([('score', {'$meta': 'textScore'})])
  501. for doc in cursor:
  502. print(doc)
  503. Raises :class:`~pymongo.errors.InvalidOperation` if this cursor has
  504. already been used. Only the last :meth:`sort` applied to this
  505. cursor has any effect.
  506. :Parameters:
  507. - `key_or_list`: a single key or a list of (key, direction)
  508. pairs specifying the keys to sort on
  509. - `direction` (optional): only used if `key_or_list` is a single
  510. key, if not given :data:`~pymongo.ASCENDING` is assumed
  511. """
  512. self.__check_okay_to_chain()
  513. keys = helpers._index_list(key_or_list, direction)
  514. self.__ordering = helpers._index_document(keys)
  515. return self
  516. def count(self, with_limit_and_skip=False):
  517. """Get the size of the results set for this query.
  518. Returns the number of documents in the results set for this query. Does
  519. not take :meth:`limit` and :meth:`skip` into account by default - set
  520. `with_limit_and_skip` to ``True`` if that is the desired behavior.
  521. Raises :class:`~pymongo.errors.OperationFailure` on a database error.
  522. When used with MongoDB >= 2.6, :meth:`~count` uses any :meth:`~hint`
  523. applied to the query. In the following example the hint is passed to
  524. the count command:
  525. collection.find({'field': 'value'}).hint('field_1').count()
  526. The :meth:`count` method obeys the
  527. :attr:`~pymongo.collection.Collection.read_preference` of the
  528. :class:`~pymongo.collection.Collection` instance on which
  529. :meth:`~pymongo.collection.Collection.find` was called.
  530. :Parameters:
  531. - `with_limit_and_skip` (optional): take any :meth:`limit` or
  532. :meth:`skip` that has been applied to this cursor into account when
  533. getting the count
  534. .. note:: The `with_limit_and_skip` parameter requires server
  535. version **>= 1.1.4-**
  536. .. versionchanged:: 2.8
  537. The :meth:`~count` method now supports :meth:`~hint`.
  538. """
  539. validate_boolean("with_limit_and_skip", with_limit_and_skip)
  540. cmd = SON([("count", self.__collection.name),
  541. ("query", self.__spec)])
  542. if self.__max_time_ms is not None:
  543. cmd["maxTimeMS"] = self.__max_time_ms
  544. if self.__comment:
  545. cmd["$comment"] = self.__comment
  546. if self.__hint is not None:
  547. cmd["hint"] = self.__hint
  548. if with_limit_and_skip:
  549. if self.__limit:
  550. cmd["limit"] = self.__limit
  551. if self.__skip:
  552. cmd["skip"] = self.__skip
  553. return self.__collection._count(cmd)
  554. def distinct(self, key):
  555. """Get a list of distinct values for `key` among all documents
  556. in the result set of this query.
  557. Raises :class:`TypeError` if `key` is not an instance of
  558. :class:`basestring` (:class:`str` in python 3).
  559. The :meth:`distinct` method obeys the
  560. :attr:`~pymongo.collection.Collection.read_preference` of the
  561. :class:`~pymongo.collection.Collection` instance on which
  562. :meth:`~pymongo.collection.Collection.find` was called.
  563. :Parameters:
  564. - `key`: name of key for which we want to get the distinct values
  565. .. seealso:: :meth:`pymongo.collection.Collection.distinct`
  566. """
  567. options = {}
  568. if self.__spec:
  569. options["query"] = self.__spec
  570. if self.__max_time_ms is not None:
  571. options['maxTimeMS'] = self.__max_time_ms
  572. if self.__comment:
  573. options['$comment'] = self.__comment
  574. return self.__collection.distinct(key, **options)
  575. def explain(self):
  576. """Returns an explain plan record for this cursor.
  577. .. mongodoc:: explain
  578. """
  579. c = self.clone()
  580. c.__explain = True
  581. # always use a hard limit for explains
  582. if c.__limit:
  583. c.__limit = -abs(c.__limit)
  584. return next(c)
  585. def hint(self, index):
  586. """Adds a 'hint', telling Mongo the proper index to use for the query.
  587. Judicious use of hints can greatly improve query
  588. performance. When doing a query on multiple fields (at least
  589. one of which is indexed) pass the indexed field as a hint to
  590. the query. Hinting will not do anything if the corresponding
  591. index does not exist. Raises
  592. :class:`~pymongo.errors.InvalidOperation` if this cursor has
  593. already been used.
  594. `index` should be an index as passed to
  595. :meth:`~pymongo.collection.Collection.create_index`
  596. (e.g. ``[('field', ASCENDING)]``) or the name of the index.
  597. If `index` is ``None`` any existing hint for this query is
  598. cleared. The last hint applied to this cursor takes precedence
  599. over all others.
  600. :Parameters:
  601. - `index`: index to hint on (as an index specifier)
  602. .. versionchanged:: 2.8
  603. The :meth:`~hint` method accepts the name of the index.
  604. """
  605. self.__check_okay_to_chain()
  606. if index is None:
  607. self.__hint = None
  608. return self
  609. if isinstance(index, string_type):
  610. self.__hint = index
  611. else:
  612. self.__hint = helpers._index_document(index)
  613. return self
  614. def comment(self, comment):
  615. """Adds a 'comment' to the cursor.
  616. http://docs.mongodb.org/manual/reference/operator/comment/
  617. :Parameters:
  618. - `comment`: A string or document
  619. .. versionadded:: 2.7
  620. """
  621. self.__check_okay_to_chain()
  622. self.__comment = comment
  623. return self
  624. def where(self, code):
  625. """Adds a $where clause to this query.
  626. The `code` argument must be an instance of :class:`basestring`
  627. (:class:`str` in python 3) or :class:`~bson.code.Code`
  628. containing a JavaScript expression. This expression will be
  629. evaluated for each document scanned. Only those documents
  630. for which the expression evaluates to *true* will be returned
  631. as results. The keyword *this* refers to the object currently
  632. being scanned.
  633. Raises :class:`TypeError` if `code` is not an instance of
  634. :class:`basestring` (:class:`str` in python 3). Raises
  635. :class:`~pymongo.errors.InvalidOperation` if this
  636. :class:`Cursor` has already been used. Only the last call to
  637. :meth:`where` applied to a :class:`Cursor` has any effect.
  638. :Parameters:
  639. - `code`: JavaScript expression to use as a filter
  640. """
  641. self.__check_okay_to_chain()
  642. if not isinstance(code, Code):
  643. code = Code(code)
  644. self.__spec["$where"] = code
  645. return self
  646. def __send_message(self, operation):
  647. """Send a query or getmore operation and handles the response.
  648. If operation is ``None`` this is an exhaust cursor, which reads
  649. the next result batch off the exhaust socket instead of
  650. sending getMore messages to the server.
  651. Can raise ConnectionFailure.
  652. """
  653. client = self.__collection.database.client
  654. if operation:
  655. kwargs = {
  656. "read_preference": self.__read_preference,
  657. "exhaust": self.__exhaust,
  658. }
  659. if self.__address is not None:
  660. kwargs["address"] = self.__address
  661. try:
  662. response = client._send_message_with_response(operation,
  663. **kwargs)
  664. self.__address = response.address
  665. if self.__exhaust:
  666. # 'response' is an ExhaustResponse.
  667. self.__exhaust_mgr = _SocketManager(response.socket_info,
  668. response.pool)
  669. data = response.data
  670. except AutoReconnect:
  671. # Don't try to send kill cursors on another socket
  672. # or to another server. It can cause a _pinValue
  673. # assertion on some server releases if we get here
  674. # due to a socket timeout.
  675. self.__killed = True
  676. raise
  677. else:
  678. # Exhaust cursor - no getMore message.
  679. try:
  680. data = self.__exhaust_mgr.sock.receive_message(1, None)
  681. except ConnectionFailure:
  682. self.__die()
  683. raise
  684. try:
  685. doc = helpers._unpack_response(response=data,
  686. cursor_id=self.__id,
  687. codec_options=self.__codec_options)
  688. except OperationFailure:
  689. self.__killed = True
  690. # Make sure exhaust socket is returned immediately, if necessary.
  691. self.__die()
  692. # If this is a tailable cursor the error is likely
  693. # due to capped collection roll over. Setting
  694. # self.__killed to True ensures Cursor.alive will be
  695. # False. No need to re-raise.
  696. if self.__query_flags & _QUERY_OPTIONS["tailable_cursor"]:
  697. return
  698. raise
  699. except NotMasterError:
  700. # Don't send kill cursors to another server after a "not master"
  701. # error. It's completely pointless.
  702. self.__killed = True
  703. # Make sure exhaust socket is returned immediately, if necessary.
  704. self.__die()
  705. client._reset_server_and_request_check(self.__address)
  706. raise
  707. self.__id = doc["cursor_id"]
  708. if self.__id == 0:
  709. self.__killed = True
  710. # starting from doesn't get set on getmore's for tailable cursors
  711. if not self.__query_flags & _QUERY_OPTIONS["tailable_cursor"]:
  712. assert doc["starting_from"] == self.__retrieved, (
  713. "Result batch started from %s, expected %s" % (
  714. doc['starting_from'], self.__retrieved))
  715. self.__retrieved += doc["number_returned"]
  716. self.__data = deque(doc["data"])
  717. if self.__limit and self.__id and self.__limit <= self.__retrieved:
  718. self.__die()
  719. # Don't wait for garbage collection to call __del__, return the
  720. # socket to the pool now.
  721. if self.__exhaust and self.__id == 0:
  722. self.__exhaust_mgr.close()
  723. def _refresh(self):
  724. """Refreshes the cursor with more data from Mongo.
  725. Returns the length of self.__data after refresh. Will exit early if
  726. self.__data is already non-empty. Raises OperationFailure when the
  727. cursor cannot be refreshed due to an error on the query.
  728. """
  729. if len(self.__data) or self.__killed:
  730. return len(self.__data)
  731. if self.__id is None: # Query
  732. ntoreturn = self.__batch_size
  733. if self.__limit:
  734. if self.__batch_size:
  735. ntoreturn = min(self.__limit, self.__batch_size)
  736. else:
  737. ntoreturn = self.__limit
  738. self.__send_message(_Query(self.__query_flags,
  739. self.__collection.full_name,
  740. self.__skip,
  741. ntoreturn,
  742. self.__query_spec(),
  743. self.__projection,
  744. self.__codec_options,
  745. self.__read_preference))
  746. if not self.__id:
  747. self.__killed = True
  748. elif self.__id: # Get More
  749. if self.__limit:
  750. limit = self.__limit - self.__retrieved
  751. if self.__batch_size:
  752. limit = min(limit, self.__batch_size)
  753. else:
  754. limit = self.__batch_size
  755. # Exhaust cursors don't send getMore messages.
  756. if self.__exhaust:
  757. self.__send_message(None)
  758. else:
  759. self.__send_message(_GetMore(self.__collection.full_name,
  760. limit,
  761. self.__id))
  762. else: # Cursor id is zero nothing else to return
  763. self.__killed = True
  764. return len(self.__data)
  765. @property
  766. def alive(self):
  767. """Does this cursor have the potential to return more data?
  768. This is mostly useful with `tailable cursors
  769. <http://www.mongodb.org/display/DOCS/Tailable+Cursors>`_
  770. since they will stop iterating even though they *may* return more
  771. results in the future.
  772. With regular cursors, simply use a for loop instead of :attr:`alive`::
  773. for doc in collection.find():
  774. print(doc)
  775. .. note:: Even if :attr:`alive` is True, :meth:`next` can raise
  776. :exc:`StopIteration`. :attr:`alive` can also be True while iterating
  777. a cursor from a failed server. In this case :attr:`alive` will
  778. return False after :meth:`next` fails to retrieve the next batch
  779. of results from the server.
  780. """
  781. return bool(len(self.__data) or (not self.__killed))
  782. @property
  783. def cursor_id(self):
  784. """Returns the id of the cursor
  785. Useful if you need to manage cursor ids and want to handle killing
  786. cursors manually using
  787. :meth:`~pymongo.mongo_client.MongoClient.kill_cursors`
  788. .. versionadded:: 2.2
  789. """
  790. return self.__id
  791. @property
  792. def address(self):
  793. """The (host, port) of the server used, or None.
  794. .. versionchanged:: 3.0
  795. Renamed from "conn_id".
  796. """
  797. return self.__address
  798. def __iter__(self):
  799. return self
  800. def next(self):
  801. """Advance the cursor."""
  802. if self.__empty:
  803. raise StopIteration
  804. _db = self.__collection.database
  805. if len(self.__data) or self._refresh():
  806. if self.__manipulate:
  807. return _db._fix_outgoing(self.__data.popleft(),
  808. self.__collection)
  809. else:
  810. return self.__data.popleft()
  811. else:
  812. raise StopIteration
  813. __next__ = next
  814. def __enter__(self):
  815. return self
  816. def __exit__(self, exc_type, exc_val, exc_tb):
  817. self.__die()
  818. def __copy__(self):
  819. """Support function for `copy.copy()`.
  820. .. versionadded:: 2.4
  821. """
  822. return self._clone(deepcopy=False)
  823. def __deepcopy__(self, memo):
  824. """Support function for `copy.deepcopy()`.
  825. .. versionadded:: 2.4
  826. """
  827. return self._clone(deepcopy=True)
  828. def _deepcopy(self, x, memo=None):
  829. """Deepcopy helper for the data dictionary or list.
  830. Regular expressions cannot be deep copied but as they are immutable we
  831. don't have to copy them when cloning.
  832. """
  833. if not hasattr(x, 'items'):
  834. y, is_list, iterator = [], True, enumerate(x)
  835. else:
  836. y, is_list, iterator = {}, False, iteritems(x)
  837. if memo is None:
  838. memo = {}
  839. val_id = id(x)
  840. if val_id in memo:
  841. return memo.get(val_id)
  842. memo[val_id] = y
  843. for key, value in iterator:
  844. if isinstance(value, (dict, list)) and not isinstance(value, SON):
  845. value = self._deepcopy(value, memo)
  846. elif not isinstance(value, RE_TYPE):
  847. value = copy.deepcopy(value, memo)
  848. if is_list:
  849. y.append(value)
  850. else:
  851. if not isinstance(key, RE_TYPE):
  852. key = copy.deepcopy(key, memo)
  853. y[key] = value
  854. return y