Source code for txmongo.filter

# Copyright 2009-2014 The TxMongo Developers.  All rights reserved.
# Use of this source code is governed by the Apache License that can be
# found in the LICENSE file.

from __future__ import absolute_import, division
from collections import defaultdict
from twisted.python.compat import unicode

"""Query filters"""

def _direction(keys, direction):
    if isinstance(keys, (bytes, unicode)):
        return (keys, direction),
    elif isinstance(keys, (list, tuple)):
        return tuple([(k, direction) for k in keys])

[docs]def ASCENDING(keys): """Ascending sort order""" return _direction(keys, 1)
[docs]def DESCENDING(keys): """Descending sort order""" return _direction(keys, -1)
[docs]def GEO2D(keys): """ Two-dimensional geospatial index """ return _direction(keys, "2d")
[docs]def GEO2DSPHERE(keys): """ Two-dimensional geospatial index """ return _direction(keys, "2dsphere")
[docs]def GEOHAYSTACK(keys): """ Bucket-based geospatial index """ return _direction(keys, "geoHaystack")
[docs]def TEXT(keys): """ Text-based index """ return _direction(keys, "text")
class _QueryFilter(defaultdict): ALLOWED_DIRECTIONS = {1, -1, "2d", "2dsphere", "geoHaystack", "text"} def __init__(self): defaultdict.__init__(self, lambda: ()) def __add__(self, obj): for k, v in obj.items(): if isinstance(v, tuple): self[k] += v else: self[k] = v return self def _index_document(self, operation, index_list): name = self.__class__.__name__ try: assert isinstance(index_list, (list, tuple)) for key, direction in index_list: if not isinstance(key, (bytes, unicode)): raise TypeError("TxMongo: invalid {0}ing key '{1}'" .format(name, repr(key))) if direction not in self.ALLOWED_DIRECTIONS: raise TypeError("TxMongo invalid {0}ing direction '{1}'" .format(name, direction)) self[operation] += tuple(((key, direction),)) except Exception: raise TypeError("TxMongo: invalid list of keys for {0}, {1}" .format(name, repr(index_list))) def __repr__(self): return "<mongodb QueryFilter: %s>" % dict.__repr__(self)
[docs]class sort(_QueryFilter): """Sorts the results of a query.""" def __init__(self, key_list): _QueryFilter.__init__(self) try: assert isinstance(key_list[0], (list, tuple)) except: key_list = (key_list,) self._index_document("orderby", key_list)
[docs]class hint(_QueryFilter): """Adds a `hint`, telling Mongo the proper index to use for the query.""" def __init__(self, index_list_or_name): _QueryFilter.__init__(self) if isinstance(index_list_or_name, (list, tuple)): if not isinstance(index_list_or_name[0], (list, tuple)): index_list_or_name = (index_list_or_name,) self._index_document("hint", index_list_or_name) else: self["hint"] = index_list_or_name
[docs]class explain(_QueryFilter): """Returns an explain plan for the query.""" def __init__(self): _QueryFilter.__init__(self) self["explain"] = True
[docs]class snapshot(_QueryFilter): def __init__(self): _QueryFilter.__init__(self) self["snapshot"] = True
[docs]class comment(_QueryFilter): def __init__(self, comment): _QueryFilter.__init__(self) self["comment"] = comment