"tailable_cursor": 2,

"slave_okay": 4,

"oplog_replay": 8,

"no_timeout": 16,

"await_data": 32,

"exhaust": 64,

"""A cursor / iterator over Mongo query results.

def __init__(self, collection, spec=None, fields=None, skip=0, limit=0,

timeout=True, snapshot=False, tailable=False, sort=None,

max_scan=None, as_class=None, slave_okay=False,

await_data=False, partial=False, manipulate=True,

read_preference=ReadPreference.PRIMARY,

_must_use_master=False, _is_command=False,

_uuid_subtype=None, **kwargs):

"""Create a new cursor.

self.__id = None

if spec is None:

spec = {}

if not isinstance(spec, dict):

raise TypeError("spec must be an instance of dict")

if not isinstance(skip, int):

raise TypeError("skip must be an instance of int")

if not isinstance(limit, int):

raise TypeError("limit must be an instance of int")

if not isinstance(timeout, bool):

raise TypeError("timeout must be an instance of bool")

if not isinstance(snapshot, bool):

raise TypeError("snapshot must be an instance of bool")

if not isinstance(tailable, bool):

raise TypeError("tailable must be an instance of bool")

if not isinstance(slave_okay, bool):

raise TypeError("slave_okay must be an instance of bool")

if not isinstance(await_data, bool):

raise TypeError("await_data must be an instance of bool")

if not isinstance(partial, bool):

raise TypeError("partial must be an instance of bool")

if fields is not None:

if not fields:

fields = {"_id": 1}

if not isinstance(fields, dict):

fields = helpers._fields_list_to_dict(fields)

if as_class is None:

as_class = collection.database.connection.document_class

self.__collection = collection

self.__spec = spec

self.__fields = fields

self.__skip = skip

self.__limit = limit

self.__batch_size = 0

# This is ugly. People want to be able to do cursor[5:5] and

# get an empty result set (old behavior was an

# exception). It's hard to do that right, though, because the

# server uses limit(0) to mean 'no limit'. So we set __empty

# in that case and check for it when iterating. We also unset

# it anytime we change __limit.

self.__empty = False

self.__timeout = timeout

self.__tailable = tailable

self.__await_data = tailable and await_data

self.__partial = partial

self.__snapshot = snapshot

self.__ordering = sort and helpers._index_document(sort) or None

self.__max_scan = max_scan

self.__explain = False

self.__hint = None

self.__as_class = as_class

self.__slave_okay = slave_okay

self.__manipulate = manipulate

self.__read_preference = read_preference

self.__tz_aware = collection.database.connection.tz_aware

self.__must_use_master = _must_use_master

self.__is_command = _is_command

self.__uuid_subtype = _uuid_subtype or collection.uuid_subtype

self.__query_flags = 0

self.__data = []

self.__connection_id = None

self.__retrieved = 0

self.__killed = False

# this is for passing network_timeout through if it's specified

# need to use kwargs as None is a legit value for network_timeout

self.__kwargs = kwargs

@property

def collection(self):

"""The :class:`~pymongo.collection.Collection` that this

return self.__collection

def __del__(self):

if self.__id and not self.__killed:

self.__die()

def rewind(self):

"""Rewind this cursor to it's unevaluated state.

self.__data = []

self.__id = None

self.__connection_id = None

self.__retrieved = 0

self.__killed = False

return self

def clone(self):

"""Get a clone of this cursor.

copy = Cursor(self.__collection, self.__spec, self.__fields,

self.__skip, self.__limit, self.__timeout,

self.__snapshot, self.__tailable)

copy.__ordering = self.__ordering

copy.__explain = self.__explain

copy.__hint = self.__hint

copy.__batch_size = self.__batch_size

copy.__max_scan = self.__max_scan

copy.__as_class = self.__as_class

copy.__slave_okay = self.__slave_okay

copy.__await_data = self.__await_data

copy.__partial = self.__partial

copy.__manipulate = self.__manipulate

copy.__read_preference = self.__read_preference

copy.__must_use_master = self.__must_use_master

copy.__is_command = self.__is_command

copy.__uuid_subtype = self.__uuid_subtype

copy.__query_flags = self.__query_flags

copy.__kwargs = self.__kwargs

return copy

def __die(self):

"""Closes this cursor.

if self.__id and not self.__killed:

connection = self.__collection.database.connection

if self.__connection_id is not None:

connection.close_cursor(self.__id, self.__connection_id)

else:

connection.close_cursor(self.__id)

self.__killed = True

def close(self):

"""Explicitly close this cursor. Required for PyPy, Jython and

self.__die()

def __query_spec(self):

"""Get the spec to use for a query.

spec = self.__spec

if not self.__is_command and "$query" not in self.__spec:

spec = SON({"$query": self.__spec})

if self.__ordering:

spec["$orderby"] = self.__ordering

if self.__explain:

spec["$explain"] = True

if self.__hint:

spec["$hint"] = self.__hint

if self.__snapshot:

spec["$snapshot"] = True

if self.__max_scan:

spec["$maxScan"] = self.__max_scan

return spec

def __query_options(self):

"""Get the query options string to use for this query.

options = self.__query_flags

if self.__tailable:

options |= _QUERY_OPTIONS["tailable_cursor"]

if self.__slave_okay or self.__read_preference:

options |= _QUERY_OPTIONS["slave_okay"]

if not self.__timeout:

options |= _QUERY_OPTIONS["no_timeout"]

if self.__await_data:

options |= _QUERY_OPTIONS["await_data"]

if self.__partial:

options |= _QUERY_OPTIONS["partial"]

return options

def __check_okay_to_chain(self):

"""Check if it is okay to chain more options onto this cursor.

if self.__retrieved or self.__id is not None:

raise InvalidOperation("cannot set options after executing query")

def add_option(self, mask):

"""Set arbitary query flags using a bitmask.

if not isinstance(mask, int):

raise TypeError("mask must be an int")

self.__check_okay_to_chain()

self.__query_flags |= mask

return self

def remove_option(self, mask):

"""Unset arbitrary query flags using a bitmask.

if not isinstance(mask, int):

raise TypeError("mask must be an int")

self.__check_okay_to_chain()

self.__query_flags &= ~mask

return self

def limit(self, limit):

"""Limits the number of results to be returned by this cursor.

if not isinstance(limit, int):

raise TypeError("limit must be an int")

self.__check_okay_to_chain()

self.__empty = False

self.__limit = limit

return self

def batch_size(self, batch_size):

"""Set the size for batches of results returned by this cursor.

if not isinstance(batch_size, int):

raise TypeError("batch_size must be an int")

if batch_size < 0:

raise ValueError("batch_size must be >= 0")

self.__check_okay_to_chain()

self.__batch_size = batch_size == 1 and 2 or batch_size

return self

def skip(self, skip):

"""Skips the first `skip` results of this cursor.

if not isinstance(skip, (int, long)):

raise TypeError("skip must be an int")

self.__check_okay_to_chain()

self.__skip = skip

return self

def __getitem__(self, index):

"""Get a single document or a slice of documents from this cursor.

self.__check_okay_to_chain()

self.__empty = False

if isinstance(index, slice):

if index.step is not None:

raise IndexError("Cursor instances do not support slice steps")

skip = 0

if index.start is not None:

if index.start < 0:

raise IndexError("Cursor instances do not support"

"negative indices")

skip = index.start

if index.stop is not None:

limit = index.stop - skip

if limit < 0:

raise IndexError("stop index must be greater than start"

"index for slice %r" % index)

if limit == 0:

self.__empty = True

else:

limit = 0

self.__skip = skip

self.__limit = limit

return self

if isinstance(index, (int, long)):

if index < 0:

raise IndexError("Cursor instances do not support negative"

"indices")

clone = self.clone()

clone.skip(index + self.__skip)

clone.limit(-1) # use a hard limit

for doc in clone:

return doc

raise IndexError("no such item for Cursor instance")

raise TypeError("index %r cannot be applied to Cursor "

"instances" % index)

def max_scan(self, max_scan):

"""Limit the number of documents to scan when performing the query.

self.__check_okay_to_chain()

self.__max_scan = max_scan

return self

def sort(self, key_or_list, direction=None):

"""Sorts this cursor's results.

self.__check_okay_to_chain()

keys = helpers._index_list(key_or_list, direction)

self.__ordering = helpers._index_document(keys)

return self

def count(self, with_limit_and_skip=False):

"""Get the size of the results set for this query.

command = {"query": self.__spec, "fields": self.__fields}

use_master = not self.__slave_okay and not self.__read_preference

command['_use_master'] = use_master

if with_limit_and_skip:

if self.__limit:

command["limit"] = self.__limit

if self.__skip:

command["skip"] = self.__skip

database = self.__collection.database

r = database.command("count", self.__collection.name,

allowable_errors=["ns missing"],

uuid_subtype = self.__uuid_subtype,

**command)

if r.get("errmsg", "") == "ns missing":

return 0

return int(r["n"])

def distinct(self, key):

"""Get a list of distinct values for `key` among all documents

if not isinstance(key, basestring):

raise TypeError("key must be an instance of basestring")

options = {"key": key}

if self.__spec:

options["query"] = self.__spec

use_master = not self.__slave_okay and not self.__read_preference

options['_use_master'] = use_master

database = self.__collection.database

return database.command("distinct",

self.__collection.name,

uuid_subtype = self.__uuid_subtype,

**options)["values"]

def explain(self):

"""Returns an explain plan record for this cursor.

c = self.clone()

c.__explain = True

# always use a hard limit for explains

if c.__limit:

c.__limit = -abs(c.__limit)

return c.next()

def hint(self, index):

"""Adds a 'hint', telling Mongo the proper index to use for the query.

self.__check_okay_to_chain()

if index is None:

self.__hint = None

return self

self.__hint = helpers._index_document(index)

return self

def where(self, code):

"""Adds a $where clause to this query.

self.__check_okay_to_chain()

if not isinstance(code, Code):

code = Code(code)

self.__spec["$where"] = code

return self

def __send_message(self, message):

"""Send a query or getmore message and handles the response.

db = self.__collection.database

kwargs = {"_must_use_master": self.__must_use_master}

kwargs["read_preference"] = self.__read_preference

if self.__connection_id is not None:

kwargs["_connection_to_use"] = self.__connection_id

kwargs.update(self.__kwargs)

response = db.connection._send_message_with_response(message,

**kwargs)

if isinstance(response, tuple):

(connection_id, response) = response

else:

connection_id = None

self.__connection_id = connection_id

try:

response = helpers._unpack_response(response, self.__id,

self.__as_class,

self.__tz_aware)

except AutoReconnect:

db.connection.disconnect()

raise

self.__id = response["cursor_id"]

# starting from doesn't get set on getmore's for tailable cursors

if not self.__tailable:

assert response["starting_from"] == self.__retrieved

self.__retrieved += response["number_returned"]

self.__data = response["data"]

if self.__limit and self.__id and self.__limit <= self.__retrieved:

self.__die()

def _refresh(self):

"""Refreshes the cursor with more data from Mongo.

if len(self.__data) or self.__killed:

return len(self.__data)

if self.__id is None: # Query

ntoreturn = self.__batch_size

if self.__limit:

if self.__batch_size:

ntoreturn = min(self.__limit, self.__batch_size)

else:

ntoreturn = self.__limit

self.__send_message(

message.query(self.__query_options(),

self.__collection.full_name,

self.__skip, ntoreturn,

self.__query_spec(), self.__fields,

self.__uuid_subtype))

if not self.__id:

self.__killed = True

elif self.__id: # Get More

if self.__limit:

limit = self.__limit - self.__retrieved

if self.__batch_size:

limit = min(limit, self.__batch_size)

else:

limit = self.__batch_size

self.__send_message(

message.get_more(self.__collection.full_name,

limit, self.__id))

return len(self.__data)

@property

def alive(self):

"""Does this cursor have the potential to return more data?

return bool(len(self.__data) or (not self.__killed))

def __iter__(self):

return self

def next(self):

if self.__empty:

raise StopIteration

db = self.__collection.database

if len(self.__data) or self._refresh():

if self.__manipulate:

return db._fix_outgoing(self.__data.pop(0), self.__collection)

else:

return self.__data.pop(0)

else:

raise StopIteration

def __enter__(self):

return self

def __exit__(self, exc_type, exc_val, exc_tb):