<s> import collections <EOL> from collections import OrderedDict <EOL> import copy <EOL> import functools <EOL> import itertools <EOL> import json <EOL> from operator import itemgetter <EOL> import time <EOL> import warnings <EOL> try : <EOL> from bson import json_util , SON <EOL> except ImportError : <EOL> json_utils = SON = None <EOL> try : <EOL> import execjs <EOL> except ImportError : <EOL> execjs = None <EOL> try : <EOL> from pymongo import ReturnDocument <EOL> except ImportError : <EOL> class ReturnDocument ( object ) : <EOL> BEFORE = False <EOL> AFTER = True <EOL> from sentinels import NOTHING <EOL> from six import iteritems <EOL> from six import iterkeys <EOL> from six import itervalues <EOL> from six import string_types <EOL> from six import text_type <EOL> from mongomock . command_cursor import CommandCursor <EOL> from mongomock import DuplicateKeyError <EOL> from mongomock . filtering import filter_applies <EOL> from mongomock . filtering import iter_key_candidates <EOL> from mongomock import helpers <EOL> from mongomock import InvalidOperation <EOL> from mongomock import ObjectId <EOL> from mongomock import OperationFailure <EOL> from mongomock . results import DeleteResult <EOL> from mongomock . results import InsertManyResult <EOL> from mongomock . results import InsertOneResult <EOL> from mongomock . results import UpdateResult <EOL> from mongomock . write_concern import WriteConcern <EOL> from mongomock import WriteError <EOL> def validate_is_mapping ( option , value ) : <EOL> if not isinstance ( value , collections . Mapping ) : <EOL> raise TypeError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % ( option , ) ) <EOL> def validate_is_mutable_mapping ( option , value ) : <EOL> if not isinstance ( value , collections . MutableMapping ) : <EOL> raise TypeError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % ( option , ) ) <EOL> def validate_ok_for_replace ( replacement ) : <EOL> validate_is_mapping ( '<STR_LIT>' , replacement ) <EOL> if replacement : <EOL> first = next ( iter ( replacement ) ) <EOL> if first . startswith ( '<STR_LIT:$>' ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> def validate_ok_for_update ( update ) : <EOL> validate_is_mapping ( '<STR_LIT>' , update ) <EOL> if not update : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> first = next ( iter ( update ) ) <EOL> if not first . startswith ( '<STR_LIT:$>' ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> def validate_write_concern_params ( ** params ) : <EOL> if params : <EOL> WriteConcern ( ** params ) <EOL> class BulkWriteOperation ( object ) : <EOL> def __init__ ( self , builder , selector , is_upsert = False ) : <EOL> self . builder = builder <EOL> self . selector = selector <EOL> self . is_upsert = is_upsert <EOL> def upsert ( self ) : <EOL> assert not self . is_upsert <EOL> return BulkWriteOperation ( self . builder , self . selector , is_upsert = True ) <EOL> def register_remove_op ( self , multi ) : <EOL> collection = self . builder . collection <EOL> selector = self . selector <EOL> def exec_remove ( ) : <EOL> op_result = collection . remove ( selector , multi = multi ) <EOL> if op_result . get ( "<STR_LIT>" ) : <EOL> return { '<STR_LIT>' : op_result . get ( '<STR_LIT:n>' ) } <EOL> err = op_result . get ( "<STR_LIT>" ) <EOL> if err : <EOL> return { "<STR_LIT>" : [ err ] } <EOL> return { } <EOL> self . builder . executors . append ( exec_remove ) <EOL> def remove ( self ) : <EOL> assert not self . is_upsert <EOL> self . register_remove_op ( multi = True ) <EOL> def remove_one ( self , ) : <EOL> assert not self . is_upsert <EOL> self . register_remove_op ( multi = False ) <EOL> def register_update_op ( self , document , multi , ** extra_args ) : <EOL> if not extra_args . get ( "<STR_LIT>" ) : <EOL> validate_ok_for_update ( document ) <EOL> collection = self . builder . collection <EOL> selector = self . selector <EOL> def exec_update ( ) : <EOL> result = collection . _update ( spec = selector , document = document , <EOL> multi = multi , upsert = self . is_upsert , <EOL> ** extra_args ) <EOL> ret_val = { } <EOL> if result . get ( '<STR_LIT>' ) : <EOL> ret_val [ "<STR_LIT>" ] = result . get ( '<STR_LIT>' ) <EOL> ret_val [ "<STR_LIT>" ] = result . get ( '<STR_LIT:n>' ) <EOL> modified = result . get ( '<STR_LIT>' ) <EOL> if modified is not None : <EOL> ret_val [ '<STR_LIT>' ] = modified <EOL> ret_val [ '<STR_LIT>' ] = modified <EOL> if result . get ( '<STR_LIT>' ) : <EOL> ret_val [ '<STR_LIT>' ] = result . get ( '<STR_LIT>' ) <EOL> return ret_val <EOL> self . builder . executors . append ( exec_update ) <EOL> def update ( self , document ) : <EOL> self . register_update_op ( document , multi = True ) <EOL> def update_one ( self , document ) : <EOL> self . register_update_op ( document , multi = False ) <EOL> def replace_one ( self , document ) : <EOL> self . register_update_op ( document , multi = False , remove = True ) <EOL> class BulkOperationBuilder ( object ) : <EOL> def __init__ ( self , collection , ordered = False ) : <EOL> self . collection = collection <EOL> self . ordered = ordered <EOL> self . results = { } <EOL> self . executors = [ ] <EOL> self . done = False <EOL> self . _insert_returns_nModified = True <EOL> self . _update_returns_nModified = True <EOL> def find ( self , selector ) : <EOL> return BulkWriteOperation ( self , selector ) <EOL> def insert ( self , doc ) : <EOL> def exec_insert ( ) : <EOL> self . collection . insert ( doc ) <EOL> return { '<STR_LIT>' : <NUM_LIT:1> } <EOL> self . executors . append ( exec_insert ) <EOL> def __aggregate_operation_result ( self , total_result , key , value ) : <EOL> agg_val = total_result . get ( key ) <EOL> assert agg_val is not None , "<STR_LIT>" "<STR_LIT>" % ( key , value ) <EOL> if isinstance ( agg_val , int ) : <EOL> total_result [ key ] += value <EOL> elif isinstance ( agg_val , list ) : <EOL> if key == "<STR_LIT>" : <EOL> new_element = { "<STR_LIT:index>" : len ( agg_val ) , "<STR_LIT>" : value } <EOL> agg_val . append ( new_element ) <EOL> else : <EOL> agg_val . append ( value ) <EOL> else : <EOL> assert False , "<STR_LIT>" "<STR_LIT>" % ( type ( agg_val ) , key , agg_val ) <EOL> def _set_nModified_policy ( self , insert , update ) : <EOL> self . _insert_returns_nModified = insert <EOL> self . _update_returns_nModified = update <EOL> def execute ( self , write_concern = None ) : <EOL> if not self . executors : <EOL> raise InvalidOperation ( "<STR_LIT>" ) <EOL> if self . done : <EOL> raise InvalidOperation ( "<STR_LIT>" ) <EOL> self . done = True <EOL> result = { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : [ ] , '<STR_LIT>' : [ ] , '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:0> } <EOL> has_update = False <EOL> has_insert = False <EOL> broken_nModified_info = False <EOL> for execute_func in self . executors : <EOL> exec_name = execute_func . __name__ <EOL> op_result = execute_func ( ) <EOL> for ( key , value ) in op_result . items ( ) : <EOL> self . __aggregate_operation_result ( result , key , value ) <EOL> if exec_name == "<STR_LIT>" : <EOL> has_update = True <EOL> if "<STR_LIT>" not in op_result : <EOL> broken_nModified_info = True <EOL> has_insert |= exec_name == "<STR_LIT>" <EOL> if broken_nModified_info : <EOL> result . pop ( '<STR_LIT>' ) <EOL> elif has_insert and self . _insert_returns_nModified : <EOL> pass <EOL> elif has_update and self . _update_returns_nModified : <EOL> pass <EOL> elif self . _update_returns_nModified and self . _insert_returns_nModified : <EOL> pass <EOL> else : <EOL> result . pop ( '<STR_LIT>' ) <EOL> return result <EOL> class Collection ( object ) : <EOL> def __init__ ( self , db , name ) : <EOL> self . name = name <EOL> self . full_name = "<STR_LIT>" . format ( db . name , name ) <EOL> self . _database = db <EOL> self . _documents = OrderedDict ( ) <EOL> self . _uniques = [ ] <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" . format ( self . _database , self . name ) <EOL> def __getitem__ ( self , name ) : <EOL> return self . _database [ self . name + '<STR_LIT:.>' + name ] <EOL> def __getattr__ ( self , name ) : <EOL> return self . __getitem__ ( name ) <EOL> def initialize_unordered_bulk_op ( self ) : <EOL> return BulkOperationBuilder ( self , ordered = False ) <EOL> def initialize_ordered_bulk_op ( self ) : <EOL> return BulkOperationBuilder ( self , ordered = True ) <EOL> def insert ( self , data , manipulate = True , check_keys = True , <EOL> continue_on_error = False , ** kwargs ) : <EOL> warnings . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" , DeprecationWarning , stacklevel = <NUM_LIT:2> ) <EOL> validate_write_concern_params ( ** kwargs ) <EOL> return self . _insert ( data ) <EOL> def insert_one ( self , document ) : <EOL> validate_is_mutable_mapping ( '<STR_LIT>' , document ) <EOL> return InsertOneResult ( self . _insert ( document ) , acknowledged = True ) <EOL> def insert_many ( self , documents , ordered = True ) : <EOL> if not isinstance ( documents , collections . Iterable ) or not documents : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> for document in documents : <EOL> validate_is_mutable_mapping ( '<STR_LIT>' , document ) <EOL> return InsertManyResult ( self . _insert ( documents ) , acknowledged = True ) <EOL> def _insert ( self , data ) : <EOL> if isinstance ( data , list ) : <EOL> return [ self . _insert ( item ) for item in data ] <EOL> if not all ( isinstance ( k , string_types ) for k in data ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if '<STR_LIT>' not in data : <EOL> data [ '<STR_LIT>' ] = ObjectId ( ) <EOL> object_id = data [ '<STR_LIT>' ] <EOL> if isinstance ( object_id , dict ) : <EOL> object_id = helpers . hashdict ( object_id ) <EOL> if object_id in self . _documents : <EOL> raise DuplicateKeyError ( "<STR_LIT>" , <NUM_LIT> ) <EOL> for unique in self . _uniques : <EOL> find_kwargs = { } <EOL> for key , direction in unique : <EOL> if key in data : <EOL> find_kwargs [ key ] = data [ key ] <EOL> answer = self . find ( find_kwargs ) <EOL> if answer . count ( ) > <NUM_LIT:0> : <EOL> raise DuplicateKeyError ( "<STR_LIT>" , <NUM_LIT> ) <EOL> self . _documents [ object_id ] = self . _internalize_dict ( data ) <EOL> return data [ '<STR_LIT>' ] <EOL> def _internalize_dict ( self , d ) : <EOL> return dict ( ( k , copy . deepcopy ( v ) ) for k , v in iteritems ( d ) ) <EOL> def _has_key ( self , doc , key ) : <EOL> key_parts = key . split ( '<STR_LIT:.>' ) <EOL> sub_doc = doc <EOL> for part in key_parts : <EOL> if part not in sub_doc : <EOL> return False <EOL> sub_doc = sub_doc [ part ] <EOL> return True <EOL> def _remove_key ( self , doc , key ) : <EOL> key_parts = key . split ( '<STR_LIT:.>' ) <EOL> sub_doc = doc <EOL> for part in key_parts [ : - <NUM_LIT:1> ] : <EOL> sub_doc = sub_doc [ part ] <EOL> del sub_doc [ key_parts [ - <NUM_LIT:1> ] ] <EOL> def update_one ( self , criteria , update , upsert = False ) : <EOL> validate_ok_for_update ( update ) <EOL> return UpdateResult ( self . _update ( criteria , update , upsert = upsert ) , <EOL> acknowledged = True ) <EOL> def update_many ( self , criteria , update , upsert = False ) : <EOL> validate_ok_for_update ( update ) <EOL> return UpdateResult ( self . _update ( criteria , update , upsert = upsert , <EOL> multi = True ) , <EOL> acknowledged = True ) <EOL> def replace_one ( self , criteria , replacement , upsert = False ) : <EOL> validate_ok_for_replace ( replacement ) <EOL> return UpdateResult ( self . _update ( criteria , replacement , upsert = upsert ) , <EOL> acknowledged = True ) <EOL> def update ( self , spec , document , upsert = False , manipulate = False , <EOL> multi = False , check_keys = False , ** kwargs ) : <EOL> warnings . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" , DeprecationWarning , stacklevel = <NUM_LIT:2> ) <EOL> return self . _update ( spec , document , upsert , manipulate , multi , <EOL> check_keys , ** kwargs ) <EOL> def _update ( self , spec , document , upsert = False , manipulate = False , <EOL> multi = False , check_keys = False , ** kwargs ) : <EOL> validate_is_mapping ( '<STR_LIT>' , spec ) <EOL> validate_is_mapping ( '<STR_LIT>' , document ) <EOL> updated_existing = False <EOL> upserted_id = None <EOL> num_updated = <NUM_LIT:0> <EOL> for existing_document in itertools . chain ( self . _iter_documents ( spec ) , [ None ] ) : <EOL> was_insert = False <EOL> if existing_document is None : <EOL> if not upsert or num_updated : <EOL> continue <EOL> _id = document . get ( '<STR_LIT>' ) <EOL> to_insert = dict ( spec , _id = _id ) if _id else spec <EOL> to_insert = self . _expand_dots ( to_insert ) <EOL> upserted_id = self . _insert ( self . _discard_operators ( to_insert ) ) <EOL> existing_document = self . _documents [ upserted_id ] <EOL> was_insert = True <EOL> else : <EOL> updated_existing = True <EOL> num_updated += <NUM_LIT:1> <EOL> first = True <EOL> subdocument = None <EOL> for k , v in iteritems ( document ) : <EOL> if k == '<STR_LIT>' : <EOL> positional = False <EOL> for key in iterkeys ( v ) : <EOL> if '<STR_LIT:$>' in key : <EOL> positional = True <EOL> break <EOL> if positional : <EOL> subdocument = self . _update_document_fields_positional ( <EOL> existing_document , v , spec , _set_updater , subdocument ) <EOL> continue <EOL> self . _update_document_fields ( existing_document , v , _set_updater ) <EOL> elif k == '<STR_LIT>' : <EOL> if not was_insert : <EOL> continue <EOL> positional = any ( '<STR_LIT:$>' in key for key in iterkeys ( v ) ) <EOL> if positional : <EOL> subdocument = self . _update_document_fields_positional ( <EOL> existing_document , v , spec , _set_updater , subdocument ) <EOL> else : <EOL> self . _update_document_fields ( existing_document , v , _set_updater ) <EOL> elif k == '<STR_LIT>' : <EOL> for field , value in iteritems ( v ) : <EOL> if self . _has_key ( existing_document , field ) : <EOL> self . _remove_key ( existing_document , field ) <EOL> elif k == '<STR_LIT>' : <EOL> positional = False <EOL> for key in iterkeys ( v ) : <EOL> if '<STR_LIT:$>' in key : <EOL> positional = True <EOL> break <EOL> if positional : <EOL> subdocument = self . _update_document_fields_positional ( <EOL> existing_document , v , spec , _inc_updater , subdocument ) <EOL> continue <EOL> self . _update_document_fields ( existing_document , v , _inc_updater ) <EOL> elif k == '<STR_LIT>' : <EOL> for field , value in iteritems ( v ) : <EOL> nested_field_list = field . rsplit ( '<STR_LIT:.>' ) <EOL> if len ( nested_field_list ) == <NUM_LIT:1> : <EOL> if field not in existing_document : <EOL> existing_document [ field ] = [ ] <EOL> if isinstance ( value , dict ) : <EOL> if '<STR_LIT>' in value : <EOL> existing_document [ field ] += [ <EOL> obj for obj in list ( value [ '<STR_LIT>' ] ) <EOL> if obj not in existing_document [ field ] ] <EOL> continue <EOL> if value not in existing_document [ field ] : <EOL> existing_document [ field ] . append ( value ) <EOL> continue <EOL> else : <EOL> subdocument = existing_document <EOL> for field in nested_field_list [ : - <NUM_LIT:1> ] : <EOL> if field not in subdocument : <EOL> subdocument [ field ] = { } <EOL> subdocument = subdocument [ field ] <EOL> push_results = [ ] <EOL> if nested_field_list [ - <NUM_LIT:1> ] in subdocument : <EOL> push_results = subdocument [ <EOL> nested_field_list [ - <NUM_LIT:1> ] ] <EOL> if isinstance ( value , dict ) and '<STR_LIT>' in value : <EOL> push_results += [ <EOL> obj for obj in list ( value [ '<STR_LIT>' ] ) <EOL> if obj not in push_results ] <EOL> elif value not in push_results : <EOL> push_results . append ( value ) <EOL> subdocument [ nested_field_list [ - <NUM_LIT:1> ] ] = push_results <EOL> elif k == '<STR_LIT>' : <EOL> for field , value in iteritems ( v ) : <EOL> nested_field_list = field . rsplit ( '<STR_LIT:.>' ) <EOL> if '<STR_LIT:$>' in nested_field_list : <EOL> if not subdocument : <EOL> subdocument = self . _get_subdocument ( <EOL> existing_document , spec , nested_field_list ) <EOL> pull_results = [ ] <EOL> for obj in subdocument [ nested_field_list [ - <NUM_LIT:1> ] ] : <EOL> if isinstance ( obj , dict ) : <EOL> for pull_key , pull_value in iteritems ( value ) : <EOL> if obj [ pull_key ] != pull_value : <EOL> pull_results . append ( obj ) <EOL> continue <EOL> if obj != value : <EOL> pull_results . append ( obj ) <EOL> subdocument [ nested_field_list [ - <NUM_LIT:1> ] ] = pull_results <EOL> else : <EOL> arr = existing_document <EOL> for field in nested_field_list : <EOL> if field not in arr : <EOL> break <EOL> arr = arr [ field ] <EOL> if not isinstance ( arr , list ) : <EOL> continue <EOL> if isinstance ( value , dict ) : <EOL> for idx , obj in enumerate ( arr ) : <EOL> if filter_applies ( value , obj ) : <EOL> del arr [ idx ] <EOL> else : <EOL> for idx , obj in enumerate ( arr ) : <EOL> if value == obj : <EOL> del arr [ idx ] <EOL> elif k == '<STR_LIT>' : <EOL> for field , value in iteritems ( v ) : <EOL> nested_field_list = field . rsplit ( '<STR_LIT:.>' ) <EOL> if len ( nested_field_list ) == <NUM_LIT:1> : <EOL> if field in existing_document : <EOL> arr = existing_document [ field ] <EOL> existing_document [ field ] = [ <EOL> obj for obj in arr if obj not in value ] <EOL> continue <EOL> else : <EOL> subdocument = existing_document <EOL> for nested_field in nested_field_list [ : - <NUM_LIT:1> ] : <EOL> if nested_field not in subdocument : <EOL> break <EOL> subdocument = subdocument [ nested_field ] <EOL> if nested_field_list [ - <NUM_LIT:1> ] in subdocument : <EOL> arr = subdocument [ nested_field_list [ - <NUM_LIT:1> ] ] <EOL> subdocument [ nested_field_list [ - <NUM_LIT:1> ] ] = [ <EOL> obj for obj in arr if obj not in value ] <EOL> elif k == '<STR_LIT>' : <EOL> for field , value in iteritems ( v ) : <EOL> nested_field_list = field . rsplit ( '<STR_LIT:.>' ) <EOL> if len ( nested_field_list ) == <NUM_LIT:1> : <EOL> if field not in existing_document : <EOL> existing_document [ field ] = [ ] <EOL> if isinstance ( value , dict ) : <EOL> if '<STR_LIT>' in value : <EOL> existing_document [ field ] += list ( value [ '<STR_LIT>' ] ) <EOL> continue <EOL> existing_document [ field ] . append ( value ) <EOL> continue <EOL> elif '<STR_LIT:$>' in nested_field_list : <EOL> if not subdocument : <EOL> subdocument = self . _get_subdocument ( <EOL> existing_document , spec , nested_field_list ) <EOL> push_results = [ ] <EOL> if nested_field_list [ - <NUM_LIT:1> ] in subdocument : <EOL> push_results = subdocument [ nested_field_list [ - <NUM_LIT:1> ] ] <EOL> if isinstance ( value , dict ) : <EOL> if '<STR_LIT>' in value : <EOL> push_results += list ( value [ '<STR_LIT>' ] ) <EOL> else : <EOL> push_results . append ( value ) <EOL> else : <EOL> push_results . append ( value ) <EOL> subdocument [ nested_field_list [ - <NUM_LIT:1> ] ] = push_results <EOL> else : <EOL> subdocument = existing_document <EOL> for field in nested_field_list [ : - <NUM_LIT:1> ] : <EOL> if field not in subdocument : <EOL> subdocument [ field ] = { } <EOL> subdocument = subdocument [ field ] <EOL> push_results = [ ] <EOL> if nested_field_list [ - <NUM_LIT:1> ] in subdocument : <EOL> push_results = subdocument [ nested_field_list [ - <NUM_LIT:1> ] ] <EOL> if isinstance ( value , dict ) and '<STR_LIT>' in value : <EOL> push_results += list ( value [ '<STR_LIT>' ] ) <EOL> else : <EOL> push_results . append ( value ) <EOL> subdocument [ nested_field_list [ - <NUM_LIT:1> ] ] = push_results <EOL> else : <EOL> if first : <EOL> for key in document . keys ( ) : <EOL> if key . startswith ( '<STR_LIT:$>' ) : <EOL> raise ValueError ( '<STR_LIT>' . format ( k ) ) <EOL> _id = spec . get ( '<STR_LIT>' , existing_document . get ( '<STR_LIT>' ) ) <EOL> existing_document . clear ( ) <EOL> if _id : <EOL> existing_document [ '<STR_LIT>' ] = _id <EOL> existing_document . update ( self . _internalize_dict ( document ) ) <EOL> if existing_document [ '<STR_LIT>' ] != _id : <EOL> raise OperationFailure ( <EOL> "<STR_LIT>" <EOL> . format ( existing_document [ '<STR_LIT>' ] , _id ) ) <EOL> break <EOL> else : <EOL> raise ValueError ( <EOL> '<STR_LIT>' . format ( k ) ) <EOL> first = False <EOL> if not multi : <EOL> break <EOL> return { <EOL> text_type ( "<STR_LIT>" ) : self . _database . client . _id , <EOL> text_type ( "<STR_LIT>" ) : None , <EOL> text_type ( "<STR_LIT:n>" ) : num_updated , <EOL> text_type ( "<STR_LIT>" ) : num_updated if updated_existing else <NUM_LIT:0> , <EOL> text_type ( "<STR_LIT>" ) : <NUM_LIT:1> , <EOL> text_type ( "<STR_LIT>" ) : upserted_id , <EOL> text_type ( "<STR_LIT>" ) : updated_existing , <EOL> } <EOL> def _get_subdocument ( self , existing_document , spec , nested_field_list ) : <EOL> """<STR_LIT>""" <EOL> doc = existing_document <EOL> subdocument = existing_document <EOL> subspec = spec <EOL> for subfield in nested_field_list : <EOL> if subfield == '<STR_LIT:$>' : <EOL> subspec = subspec [ '<STR_LIT>' ] <EOL> for item in doc : <EOL> if filter_applies ( subspec , item ) : <EOL> subdocument = doc <EOL> doc = item <EOL> break <EOL> continue <EOL> subdocument = doc <EOL> doc = doc [ subfield ] <EOL> if subfield not in subspec : <EOL> break <EOL> subspec = subspec [ subfield ] <EOL> return subdocument <EOL> def _expand_dots ( self , doc ) : <EOL> expanded = { } <EOL> paths = { } <EOL> for k , v in iteritems ( doc ) : <EOL> key_parts = k . split ( '<STR_LIT:.>' ) <EOL> sub_doc = v <EOL> for i in reversed ( range ( <NUM_LIT:1> , len ( key_parts ) ) ) : <EOL> key = key_parts [ i ] <EOL> sub_doc = { key : sub_doc } <EOL> key = key_parts [ <NUM_LIT:0> ] <EOL> if key in expanded : <EOL> raise WriteError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % ( k , paths [ key ] ) ) <EOL> paths [ key ] = k <EOL> expanded [ key ] = sub_doc <EOL> return expanded <EOL> def _discard_operators ( self , doc ) : <EOL> return dict ( ( k , v ) for k , v in iteritems ( doc ) if not k . startswith ( "<STR_LIT:$>" ) ) <EOL> def find ( self , filter = None , projection = None , skip = <NUM_LIT:0> , limit = <NUM_LIT:0> , <EOL> no_cursor_timeout = False , cursor_type = None , sort = None , <EOL> allow_partial_results = False , oplog_replay = False , modifiers = None , <EOL> batch_size = <NUM_LIT:0> , manipulate = True ) : <EOL> spec = filter <EOL> if spec is None : <EOL> spec = { } <EOL> validate_is_mapping ( '<STR_LIT>' , spec ) <EOL> return Cursor ( self , functools . partial ( <EOL> self . _get_dataset , spec , sort , projection , dict , skip ) , limit = limit ) <EOL> def _get_dataset ( self , spec , sort , fields , as_class , skip ) : <EOL> dataset = ( self . _copy_only_fields ( document , fields , as_class ) <EOL> for document in self . _iter_documents ( spec ) ) <EOL> if sort : <EOL> for sortKey , sortDirection in reversed ( sort ) : <EOL> dataset = iter ( sorted ( <EOL> dataset , key = lambda x : _resolve_sort_key ( sortKey , x ) , <EOL> reverse = sortDirection < <NUM_LIT:0> ) ) <EOL> for i in helpers . xrange ( skip ) : <EOL> try : <EOL> next ( dataset ) <EOL> except StopIteration : <EOL> pass <EOL> return dataset <EOL> def _copy_field ( self , obj , container ) : <EOL> if isinstance ( obj , list ) : <EOL> new = [ ] <EOL> for item in obj : <EOL> new . append ( self . _copy_field ( item , container ) ) <EOL> return new <EOL> if isinstance ( obj , dict ) : <EOL> new = container ( ) <EOL> for key , value in obj . items ( ) : <EOL> new [ key ] = self . _copy_field ( value , container ) <EOL> return new <EOL> else : <EOL> return copy . copy ( obj ) <EOL> def _extract_projection_operators ( self , fields ) : <EOL> """<STR_LIT>""" <EOL> result = { } <EOL> allowed_projection_operators = set ( [ '<STR_LIT>' ] ) <EOL> for key , value in iteritems ( fields ) : <EOL> if isinstance ( value , dict ) : <EOL> for op in value : <EOL> if op not in allowed_projection_operators : <EOL> raise ValueError ( '<STR_LIT>' . format ( op ) ) <EOL> result [ key ] = value <EOL> for key in result : <EOL> del fields [ key ] <EOL> return result <EOL> def _apply_projection_operators ( self , ops , doc , doc_copy ) : <EOL> """<STR_LIT>""" <EOL> for field , op in iteritems ( ops ) : <EOL> if field not in doc_copy : <EOL> if field in doc : <EOL> doc_copy [ field ] = doc [ field ] <EOL> else : <EOL> continue <EOL> if '<STR_LIT>' in op : <EOL> if isinstance ( doc_copy [ field ] , list ) : <EOL> matched = False <EOL> for item in doc_copy [ field ] : <EOL> if filter_applies ( op [ '<STR_LIT>' ] , item ) : <EOL> matched = True <EOL> doc_copy [ field ] = [ item ] <EOL> break <EOL> if not matched : <EOL> del doc_copy [ field ] <EOL> else : <EOL> del doc_copy [ field ] <EOL> def _copy_only_fields ( self , doc , fields , container ) : <EOL> """<STR_LIT>""" <EOL> if fields is None : <EOL> return self . _copy_field ( doc , container ) <EOL> else : <EOL> if not fields : <EOL> fields = { "<STR_LIT>" : <NUM_LIT:1> } <EOL> if not isinstance ( fields , dict ) : <EOL> fields = helpers . _fields_list_to_dict ( fields ) <EOL> id_value = fields . pop ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> projection_operators = self . _extract_projection_operators ( fields ) <EOL> if len ( set ( list ( fields . values ( ) ) ) ) > <NUM_LIT:1> : <EOL> raise ValueError ( <EOL> '<STR_LIT>' ) <EOL> if len ( list ( fields . values ( ) ) ) == <NUM_LIT:0> : <EOL> if id_value == <NUM_LIT:1> : <EOL> doc_copy = container ( ) <EOL> else : <EOL> doc_copy = self . _copy_field ( doc , container ) <EOL> elif list ( fields . values ( ) ) [ <NUM_LIT:0> ] == <NUM_LIT:1> : <EOL> doc_copy = container ( ) <EOL> for key in fields : <EOL> key_parts = key . split ( '<STR_LIT:.>' ) <EOL> subdocument = doc <EOL> subdocument_copy = doc_copy <EOL> full_key_path_found = True <EOL> for key_part in key_parts [ : - <NUM_LIT:1> ] : <EOL> if key_part not in subdocument : <EOL> full_key_path_found = False <EOL> break <EOL> subdocument = subdocument [ key_part ] <EOL> subdocument_copy = subdocument_copy . setdefault ( key_part , { } ) <EOL> if not full_key_path_found or key_parts [ - <NUM_LIT:1> ] not in subdocument : <EOL> continue <EOL> subdocument_copy [ key_parts [ - <NUM_LIT:1> ] ] = subdocument [ key_parts [ - <NUM_LIT:1> ] ] <EOL> else : <EOL> doc_copy = self . _copy_field ( doc , container ) <EOL> for key in fields : <EOL> key_parts = key . split ( '<STR_LIT:.>' ) <EOL> subdocument_copy = doc_copy <EOL> full_key_path_found = True <EOL> for key_part in key_parts [ : - <NUM_LIT:1> ] : <EOL> if key_part not in subdocument_copy : <EOL> full_key_path_found = False <EOL> break <EOL> subdocument_copy = subdocument_copy [ key_part ] <EOL> if not full_key_path_found or key_parts [ - <NUM_LIT:1> ] not in subdocument_copy : <EOL> continue <EOL> del subdocument_copy [ key_parts [ - <NUM_LIT:1> ] ] <EOL> if id_value == <NUM_LIT:0> : <EOL> doc_copy . pop ( '<STR_LIT>' , None ) <EOL> else : <EOL> if '<STR_LIT>' in doc : <EOL> doc_copy [ '<STR_LIT>' ] = doc [ '<STR_LIT>' ] <EOL> fields [ '<STR_LIT>' ] = id_value <EOL> self . _apply_projection_operators ( projection_operators , doc , doc_copy ) <EOL> for field , op in iteritems ( projection_operators ) : <EOL> fields [ field ] = op <EOL> return doc_copy <EOL> def _update_document_fields ( self , doc , fields , updater ) : <EOL> """<STR_LIT>""" <EOL> for k , v in iteritems ( fields ) : <EOL> self . _update_document_single_field ( doc , k , v , updater ) <EOL> def _update_document_fields_positional ( self , doc , fields , spec , updater , <EOL> subdocument = None ) : <EOL> """<STR_LIT>""" <EOL> for k , v in iteritems ( fields ) : <EOL> if '<STR_LIT:$>' in k : <EOL> field_name_parts = k . split ( '<STR_LIT:.>' ) <EOL> if not subdocument : <EOL> current_doc = doc <EOL> subspec = spec <EOL> for part in field_name_parts [ : - <NUM_LIT:1> ] : <EOL> if part == '<STR_LIT:$>' : <EOL> subspec = subspec . get ( '<STR_LIT>' , subspec ) <EOL> for item in current_doc : <EOL> if filter_applies ( subspec , item ) : <EOL> current_doc = item <EOL> break <EOL> continue <EOL> new_spec = { } <EOL> for el in subspec : <EOL> if el . startswith ( part ) : <EOL> if len ( el . split ( "<STR_LIT:.>" ) ) > <NUM_LIT:1> : <EOL> new_spec [ "<STR_LIT:.>" . join ( <EOL> el . split ( "<STR_LIT:.>" ) [ <NUM_LIT:1> : ] ) ] = subspec [ el ] <EOL> else : <EOL> new_spec = subspec [ el ] <EOL> subspec = new_spec <EOL> current_doc = current_doc [ part ] <EOL> subdocument = current_doc <EOL> if ( field_name_parts [ - <NUM_LIT:1> ] == '<STR_LIT:$>' and <EOL> isinstance ( subdocument , list ) ) : <EOL> for i , doc in enumerate ( subdocument ) : <EOL> if filter_applies ( subspec , doc ) : <EOL> subdocument [ i ] = v <EOL> break <EOL> continue <EOL> updater ( subdocument , field_name_parts [ - <NUM_LIT:1> ] , v ) <EOL> continue <EOL> self . _update_document_single_field ( doc , k , v , updater ) <EOL> return subdocument <EOL> def _update_document_single_field ( self , doc , field_name , field_value , updater ) : <EOL> field_name_parts = field_name . split ( "<STR_LIT:.>" ) <EOL> for part in field_name_parts [ : - <NUM_LIT:1> ] : <EOL> if isinstance ( doc , list ) : <EOL> try : <EOL> if part == '<STR_LIT:$>' : <EOL> doc = doc [ <NUM_LIT:0> ] <EOL> else : <EOL> doc = doc [ int ( part ) ] <EOL> continue <EOL> except ValueError : <EOL> pass <EOL> elif isinstance ( doc , dict ) : <EOL> doc = doc . setdefault ( part , { } ) <EOL> else : <EOL> return <EOL> field_name = field_name_parts [ - <NUM_LIT:1> ] <EOL> if isinstance ( doc , list ) : <EOL> try : <EOL> doc [ int ( field_name ) ] = field_value <EOL> except IndexError : <EOL> pass <EOL> else : <EOL> updater ( doc , field_name , field_value ) <EOL> def _iter_documents ( self , filter = None ) : <EOL> return ( document for document in list ( itervalues ( self . _documents ) ) <EOL> if filter_applies ( filter , document ) ) <EOL> def find_one ( self , spec_or_id = None , * args , ** kwargs ) : <EOL> if spec_or_id is None : <EOL> spec_or_id = { } <EOL> if not isinstance ( spec_or_id , collections . Mapping ) : <EOL> spec_or_id = { '<STR_LIT>' : spec_or_id } <EOL> try : <EOL> return next ( self . find ( spec_or_id , * args , ** kwargs ) ) <EOL> except StopIteration : <EOL> return None <EOL> def find_one_and_delete ( self , filter , projection = None , sort = None , ** kwargs ) : <EOL> kwargs [ '<STR_LIT>' ] = True <EOL> validate_is_mapping ( '<STR_LIT>' , filter ) <EOL> return self . _find_and_modify ( filter , projection , sort = sort , ** kwargs ) <EOL> def find_one_and_replace ( self , filter , replacement , <EOL> projection = None , sort = None , upsert = False , <EOL> return_document = ReturnDocument . BEFORE , ** kwargs ) : <EOL> validate_is_mapping ( '<STR_LIT>' , filter ) <EOL> validate_ok_for_replace ( replacement ) <EOL> return self . _find_and_modify ( filter , projection , replacement , upsert , <EOL> sort , return_document , ** kwargs ) <EOL> def find_one_and_update ( self , filter , update , <EOL> projection = None , sort = None , upsert = False , <EOL> return_document = ReturnDocument . BEFORE , ** kwargs ) : <EOL> validate_is_mapping ( '<STR_LIT>' , filter ) <EOL> validate_ok_for_update ( update ) <EOL> return self . _find_and_modify ( filter , projection , update , upsert , <EOL> sort , return_document , ** kwargs ) <EOL> def find_and_modify ( self , query = { } , update = None , upsert = False , sort = None , <EOL> full_response = False , manipulate = False , ** kwargs ) : <EOL> warnings . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> DeprecationWarning , stacklevel = <NUM_LIT:2> ) <EOL> return self . _find_and_modify ( query , update = update , upsert = upsert , <EOL> sort = sort , ** kwargs ) <EOL> def _find_and_modify ( self , query , projection = None , update = None , <EOL> upsert = False , sort = None , <EOL> return_document = ReturnDocument . BEFORE , ** kwargs ) : <EOL> remove = kwargs . get ( "<STR_LIT>" , False ) <EOL> if kwargs . get ( "<STR_LIT>" , False ) and remove : <EOL> raise OperationFailure ( "<STR_LIT>" ) <EOL> if not ( remove or update ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if remove and update : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> old = self . find_one ( query , projection = projection , sort = sort ) <EOL> if not old and not upsert : <EOL> return <EOL> if old and '<STR_LIT>' in old : <EOL> query = { '<STR_LIT>' : old [ '<STR_LIT>' ] } <EOL> if remove : <EOL> self . delete_one ( query ) <EOL> else : <EOL> self . _update ( query , update , upsert ) <EOL> if return_document is ReturnDocument . AFTER or kwargs . get ( '<STR_LIT>' ) : <EOL> return self . find_one ( query , projection ) <EOL> return old <EOL> def save ( self , to_save , manipulate = True , check_keys = True , ** kwargs ) : <EOL> warnings . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" , DeprecationWarning , stacklevel = <NUM_LIT:2> ) <EOL> validate_is_mutable_mapping ( "<STR_LIT>" , to_save ) <EOL> validate_write_concern_params ( ** kwargs ) <EOL> if "<STR_LIT>" not in to_save : <EOL> return self . insert ( to_save ) <EOL> else : <EOL> self . _update ( { "<STR_LIT>" : to_save [ "<STR_LIT>" ] } , to_save , True , <EOL> manipulate , check_keys = True , ** kwargs ) <EOL> return to_save . get ( "<STR_LIT>" , None ) <EOL> def delete_one ( self , filter ) : <EOL> validate_is_mapping ( '<STR_LIT>' , filter ) <EOL> return DeleteResult ( self . _delete ( filter ) , True ) <EOL> def delete_many ( self , filter ) : <EOL> validate_is_mapping ( '<STR_LIT>' , filter ) <EOL> return DeleteResult ( self . _delete ( filter , multi = True ) , True ) <EOL> def _delete ( self , filter , multi = False ) : <EOL> if filter is None : <EOL> filter = { } <EOL> if not isinstance ( filter , collections . Mapping ) : <EOL> filter = { '<STR_LIT>' : filter } <EOL> to_delete = list ( self . find ( filter ) ) <EOL> deleted_count = <NUM_LIT:0> <EOL> for doc in to_delete : <EOL> doc_id = doc [ '<STR_LIT>' ] <EOL> if isinstance ( doc_id , dict ) : <EOL> doc_id = helpers . hashdict ( doc_id ) <EOL> del self . _documents [ doc_id ] <EOL> deleted_count += <NUM_LIT:1> <EOL> if not multi : <EOL> break <EOL> return { <EOL> "<STR_LIT>" : self . _database . client . _id , <EOL> "<STR_LIT:n>" : deleted_count , <EOL> "<STR_LIT>" : <NUM_LIT:1.0> , <EOL> "<STR_LIT>" : None , <EOL> } <EOL> def remove ( self , spec_or_id = None , multi = True , ** kwargs ) : <EOL> warnings . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" , DeprecationWarning , stacklevel = <NUM_LIT:2> ) <EOL> validate_write_concern_params ( ** kwargs ) <EOL> return self . _delete ( spec_or_id , multi = multi ) <EOL> def count ( self , filter = None , ** kwargs ) : <EOL> if filter is None : <EOL> return len ( self . _documents ) <EOL> else : <EOL> return self . find ( filter ) . count ( ) <EOL> def drop ( self ) : <EOL> del self . _documents <EOL> self . _documents = { } <EOL> def ensure_index ( self , key_or_list , cache_for = <NUM_LIT> , ** kwargs ) : <EOL> self . create_index ( key_or_list , cache_for , ** kwargs ) <EOL> def create_index ( self , key_or_list , cache_for = <NUM_LIT> , ** kwargs ) : <EOL> if '<STR_LIT>' in kwargs and kwargs [ '<STR_LIT>' ] : <EOL> self . _uniques . append ( helpers . _index_list ( key_or_list ) ) <EOL> def drop_index ( self , index_or_name ) : <EOL> pass <EOL> def index_information ( self ) : <EOL> return { } <EOL> def map_reduce ( self , map_func , reduce_func , out , full_response = False , <EOL> query = None , limit = <NUM_LIT:0> ) : <EOL> if execjs is None : <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> if limit == <NUM_LIT:0> : <EOL> limit = None <EOL> start_time = time . clock ( ) <EOL> out_collection = None <EOL> reduced_rows = None <EOL> full_dict = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:input>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> } , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:1.0> , <EOL> '<STR_LIT:result>' : None } <EOL> map_ctx = execjs . compile ( """<STR_LIT>""" ) <EOL> reduce_ctx = execjs . compile ( """<STR_LIT>""" ) <EOL> doc_list = [ json . dumps ( doc , default = json_util . default ) <EOL> for doc in self . find ( query ) ] <EOL> mapped_rows = map_ctx . call ( '<STR_LIT>' , map_func , doc_list ) <EOL> reduced_rows = reduce_ctx . call ( '<STR_LIT>' , reduce_func , mapped_rows ) [ : limit ] <EOL> for reduced_row in reduced_rows : <EOL> if reduced_row [ '<STR_LIT>' ] . startswith ( '<STR_LIT>' ) : <EOL> reduced_row [ '<STR_LIT>' ] = ObjectId ( reduced_row [ '<STR_LIT>' ] [ <NUM_LIT:4> : ] ) <EOL> reduced_rows = sorted ( reduced_rows , key = lambda x : x [ '<STR_LIT>' ] ) <EOL> if full_response : <EOL> full_dict [ '<STR_LIT>' ] [ '<STR_LIT:input>' ] = len ( doc_list ) <EOL> for key in mapped_rows . keys ( ) : <EOL> emit_count = len ( mapped_rows [ key ] ) <EOL> full_dict [ '<STR_LIT>' ] [ '<STR_LIT>' ] += emit_count <EOL> if emit_count > <NUM_LIT:1> : <EOL> full_dict [ '<STR_LIT>' ] [ '<STR_LIT>' ] += <NUM_LIT:1> <EOL> full_dict [ '<STR_LIT>' ] [ '<STR_LIT>' ] = len ( reduced_rows ) <EOL> if isinstance ( out , ( str , bytes ) ) : <EOL> out_collection = getattr ( self . _database , out ) <EOL> out_collection . drop ( ) <EOL> out_collection . insert ( reduced_rows ) <EOL> ret_val = out_collection <EOL> full_dict [ '<STR_LIT:result>' ] = out <EOL> elif isinstance ( out , SON ) and out . get ( '<STR_LIT:replace>' ) and out . get ( '<STR_LIT>' ) : <EOL> out_db = getattr ( self . _database . _client , out [ '<STR_LIT>' ] ) <EOL> out_collection = getattr ( out_db , out [ '<STR_LIT:replace>' ] ) <EOL> out_collection . insert ( reduced_rows ) <EOL> ret_val = out_collection <EOL> full_dict [ '<STR_LIT:result>' ] = { '<STR_LIT>' : out [ '<STR_LIT>' ] , '<STR_LIT>' : out [ '<STR_LIT:replace>' ] } <EOL> elif isinstance ( out , dict ) and out . get ( '<STR_LIT>' ) : <EOL> ret_val = reduced_rows <EOL> full_dict [ '<STR_LIT:result>' ] = reduced_rows <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> full_dict [ '<STR_LIT>' ] = int ( round ( ( time . clock ( ) - start_time ) * <NUM_LIT:1000> ) ) <EOL> if full_response : <EOL> ret_val = full_dict <EOL> return ret_val <EOL> def inline_map_reduce ( self , map_func , reduce_func , full_response = False , <EOL> query = None , limit = <NUM_LIT:0> ) : <EOL> return self . map_reduce ( <EOL> map_func , reduce_func , { '<STR_LIT>' : <NUM_LIT:1> } , full_response , query , limit ) <EOL> def distinct ( self , key ) : <EOL> return self . find ( ) . distinct ( key ) <EOL> def group ( self , key , condition , initial , reduce , finalize = None ) : <EOL> if execjs is None : <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> reduce_ctx = execjs . compile ( """<STR_LIT>""" ) <EOL> ret_array = [ ] <EOL> doc_list_copy = [ ] <EOL> ret_array_copy = [ ] <EOL> reduced_val = { } <EOL> doc_list = [ doc for doc in self . find ( condition ) ] <EOL> for doc in doc_list : <EOL> doc_copy = copy . deepcopy ( doc ) <EOL> for k in doc : <EOL> if isinstance ( doc [ k ] , ObjectId ) : <EOL> doc_copy [ k ] = str ( doc [ k ] ) <EOL> if k not in key and k not in reduce : <EOL> del doc_copy [ k ] <EOL> for initial_key in initial : <EOL> if initial_key in doc . keys ( ) : <EOL> pass <EOL> else : <EOL> doc_copy [ initial_key ] = initial [ initial_key ] <EOL> doc_list_copy . append ( doc_copy ) <EOL> doc_list = doc_list_copy <EOL> for k in key : <EOL> doc_list = sorted ( doc_list , key = lambda x : _resolve_key ( k , x ) ) <EOL> for k in key : <EOL> if not isinstance ( k , helpers . basestring ) : <EOL> raise TypeError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % helpers . basestring . __name__ ) <EOL> for k2 , group in itertools . groupby ( doc_list , lambda item : item [ k ] ) : <EOL> group_list = ( [ x for x in group ] ) <EOL> reduced_val = reduce_ctx . call ( '<STR_LIT>' , reduce , group_list ) <EOL> ret_array . append ( reduced_val ) <EOL> for doc in ret_array : <EOL> doc_copy = copy . deepcopy ( doc ) <EOL> for k in doc : <EOL> if k not in key and k not in initial . keys ( ) : <EOL> del doc_copy [ k ] <EOL> ret_array_copy . append ( doc_copy ) <EOL> ret_array = ret_array_copy <EOL> return ret_array <EOL> def _get_group_id_fields ( self , keys , parent = '<STR_LIT>' ) : <EOL> date_operators = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> fields = [ ] <EOL> if isinstance ( keys , dict ) : <EOL> for k , v in keys . items ( ) : <EOL> if isinstance ( v , str ) : <EOL> if k in date_operators : <EOL> fields . append ( ( parent + '<STR_LIT>' + k + '<STR_LIT:_>' + v ) . replace ( '<STR_LIT:$>' , '<STR_LIT>' ) ) <EOL> else : <EOL> fields . append ( v . replace ( '<STR_LIT:$>' , '<STR_LIT>' ) ) <EOL> elif isinstance ( v , dict ) : <EOL> fields . extend ( self . _get_group_id_fields ( v , parent + '<STR_LIT>' + k if parent else k ) ) <EOL> elif isinstance ( keys , str ) : <EOL> fields . append ( keys . replace ( '<STR_LIT:$>' , '<STR_LIT>' ) ) <EOL> return fields <EOL> def _add_group_id_fields ( self , out_collection , group_func_keys ) : <EOL> date_operators = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> clear_group_func_keys = [ ] <EOL> for key in group_func_keys : <EOL> out_field = key . split ( '<STR_LIT>' ) [ <NUM_LIT:0> ] <EOL> for doc in out_collection : <EOL> if key not in doc . keys ( ) : <EOL> func_field = key . split ( '<STR_LIT>' ) [ <NUM_LIT:1> ] <EOL> func , in_field = func_field . split ( '<STR_LIT:_>' ) <EOL> out_value = doc . get ( in_field ) <EOL> if func in date_operators : <EOL> if func == '<STR_LIT>' : <EOL> out_value = out_value . timetuple ( ) . tm_yday <EOL> elif func == '<STR_LIT>' : <EOL> out_value = out_value . day <EOL> elif func == '<STR_LIT>' : <EOL> out_value = out_value . isoweekday ( ) <EOL> elif func == '<STR_LIT>' : <EOL> out_value = out_value . year <EOL> elif func == '<STR_LIT>' : <EOL> out_value = out_value . month <EOL> elif func == '<STR_LIT>' : <EOL> out_value = out_value . isocalendar ( ) [ <NUM_LIT:1> ] <EOL> elif func == '<STR_LIT>' : <EOL> out_value = out_value . hour <EOL> elif func == '<STR_LIT>' : <EOL> out_value = out_value . minute <EOL> elif func == '<STR_LIT>' : <EOL> out_value = out_value . second <EOL> elif func == '<STR_LIT>' : <EOL> out_value = int ( out_value . microsecond / <NUM_LIT:1000> ) <EOL> doc [ out_field ] = out_value <EOL> clear_group_func_keys . append ( out_field ) <EOL> return out_collection , clear_group_func_keys <EOL> def aggregate ( self , pipeline , ** kwargs ) : <EOL> pipeline_operators = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> group_operators = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> boolean_operators = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> set_operators = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> compairison_operators = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> aritmetic_operators = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> string_operators = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> text_search_operators = [ '<STR_LIT>' ] <EOL> array_operators = [ '<STR_LIT>' ] <EOL> projection_operators = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> date_operators = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> conditional_operators = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> out_collection = [ doc for doc in self . find ( ) ] <EOL> grouped_collection = [ ] <EOL> for expression in pipeline : <EOL> for k , v in iteritems ( expression ) : <EOL> if k == '<STR_LIT>' : <EOL> out_collection = [ doc for doc in out_collection <EOL> if filter_applies ( v , doc ) ] <EOL> elif k == '<STR_LIT>' : <EOL> _id = expression [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> group_func_keys = self . _get_group_id_fields ( _id ) <EOL> out_collection , group_func_keys = self . _add_group_id_fields ( out_collection , <EOL> group_func_keys ) <EOL> out_collection = sorted ( out_collection , key = itemgetter ( * group_func_keys ) ) <EOL> for field , value in iteritems ( v ) : <EOL> if field != '<STR_LIT>' : <EOL> for func , key in iteritems ( value ) : <EOL> if func == "<STR_LIT>" or "<STR_LIT>" : <EOL> grouped = itertools . groupby ( out_collection , <EOL> itemgetter ( * group_func_keys ) ) <EOL> for ret_value , group in grouped : <EOL> group_list = ( [ x for x in group ] ) <EOL> ret_value = ret_value if isinstance ( ret_value , tuple ) else [ ret_value ] <EOL> doc_id = { k : v for ( k , v ) in zip ( _id . keys ( ) , ret_value ) } if isinstance ( _id , dict ) else ret_value [ <NUM_LIT:0> ] <EOL> doc_dict = { '<STR_LIT>' : doc_id } <EOL> new_doc = True <EOL> for doc in grouped_collection : <EOL> if doc [ '<STR_LIT>' ] == doc_id : <EOL> doc_dict = doc <EOL> new_doc = False <EOL> break <EOL> current_val = doc_dict . get ( field , <NUM_LIT:0> ) <EOL> from_field = key . replace ( '<STR_LIT:$>' , '<STR_LIT>' ) <EOL> if func == "<STR_LIT>" : <EOL> for doc in group_list : <EOL> current_val = sum ( [ current_val , <EOL> doc . get ( from_field , <NUM_LIT:0> ) ] ) <EOL> doc_dict [ field ] = current_val <EOL> elif func == "<STR_LIT>" : <EOL> for doc in group_list : <EOL> current_val = sum ( [ current_val , <EOL> doc . get ( from_field , <NUM_LIT:0> ) ] ) <EOL> current_avg = current_val / max ( len ( group_list ) , <NUM_LIT:1> ) <EOL> doc_dict [ field ] = current_avg <EOL> if new_doc : <EOL> grouped_collection . append ( doc_dict ) <EOL> else : <EOL> if func in group_operators : <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % func ) <EOL> else : <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % func ) <EOL> out_collection = grouped_collection <EOL> elif k == '<STR_LIT>' : <EOL> sort_array = [ ] <EOL> for x , y in v . items ( ) : <EOL> sort_array . append ( { x : y } ) <EOL> for sort_pair in reversed ( sort_array ) : <EOL> for sortKey , sortDirection in sort_pair . items ( ) : <EOL> out_collection = sorted ( <EOL> out_collection , <EOL> key = lambda x : _resolve_sort_key ( sortKey , x ) , <EOL> reverse = sortDirection < <NUM_LIT:0> ) <EOL> elif k == '<STR_LIT>' : <EOL> out_collection = out_collection [ v : ] <EOL> elif k == '<STR_LIT>' : <EOL> out_collection = out_collection [ : v ] <EOL> elif k == '<STR_LIT>' : <EOL> if not isinstance ( v , helpers . basestring ) and v [ <NUM_LIT:0> ] != '<STR_LIT:$>' : <EOL> raise ValueError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % v ) <EOL> if len ( v . split ( '<STR_LIT:.>' ) ) > <NUM_LIT:1> : <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % v ) <EOL> unwound_collection = [ ] <EOL> for doc in out_collection : <EOL> array_value = doc . get ( v [ <NUM_LIT:1> : ] ) <EOL> if array_value in ( None , [ ] ) : <EOL> continue <EOL> elif not isinstance ( array_value , list ) : <EOL> raise TypeError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % ( v , array_value ) ) <EOL> for field_item in array_value : <EOL> unwound_collection . append ( copy . deepcopy ( doc ) ) <EOL> unwound_collection [ - <NUM_LIT:1> ] [ v [ <NUM_LIT:1> : ] ] = field_item <EOL> out_collection = unwound_collection <EOL> else : <EOL> if k in pipeline_operators : <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % k ) <EOL> else : <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % k ) <EOL> return CommandCursor ( out_collection ) <EOL> def with_options ( <EOL> self , codec_options = None , read_preference = None , write_concern = None , read_concern = None ) : <EOL> return self <EOL> def _resolve_key ( key , doc ) : <EOL> return next ( iter ( iter_key_candidates ( key , doc ) ) , NOTHING ) <EOL> def _resolve_sort_key ( key , doc ) : <EOL> value = _resolve_key ( key , doc ) <EOL> if value is NOTHING : <EOL> return <NUM_LIT:0> , value <EOL> return <NUM_LIT:1> , value <EOL> class Cursor ( object ) : <EOL> def __init__ ( self , collection , dataset_factory , limit = <NUM_LIT:0> ) : <EOL> super ( Cursor , self ) . __init__ ( ) <EOL> self . collection = collection <EOL> self . _factory = dataset_factory <EOL> self . _dataset = self . _factory ( ) <EOL> self . _limit = limit if limit != <NUM_LIT:0> else None <EOL> self . _skip = None <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def clone ( self ) : <EOL> return Cursor ( self . collection , self . _factory , self . _limit ) <EOL> def __next__ ( self ) : <EOL> if self . _skip : <EOL> for i in range ( self . _skip ) : <EOL> next ( self . _dataset ) <EOL> self . _skip = None <EOL> if self . _limit is not None and self . _limit <= <NUM_LIT:0> : <EOL> raise StopIteration ( ) <EOL> if self . _limit is not None : <EOL> self . _limit -= <NUM_LIT:1> <EOL> return next ( self . _dataset ) <EOL> next = __next__ <EOL> def sort ( self , key_or_list , direction = None ) : <EOL> if direction is None : <EOL> direction = <NUM_LIT:1> <EOL> if isinstance ( key_or_list , ( tuple , list ) ) : <EOL> for sortKey , sortDirection in reversed ( key_or_list ) : <EOL> self . _dataset = iter ( <EOL> sorted ( <EOL> self . _dataset , <EOL> key = lambda x : _resolve_sort_key ( <EOL> sortKey , <EOL> x ) , <EOL> reverse = sortDirection < <NUM_LIT:0> ) ) <EOL> else : <EOL> self . _dataset = iter ( <EOL> sorted ( self . _dataset , <EOL> key = lambda x : _resolve_sort_key ( key_or_list , x ) , <EOL> reverse = direction < <NUM_LIT:0> ) ) <EOL> return self <EOL> def count ( self , with_limit_and_skip = False ) : <EOL> arr = [ x for x in self . _dataset ] <EOL> count = len ( arr ) <EOL> if with_limit_and_skip : <EOL> if self . _skip : <EOL> count -= self . _skip <EOL> if self . _limit and count > self . _limit : <EOL> count = self . _limit <EOL> self . _dataset = iter ( arr ) <EOL> return count <EOL> def skip ( self , count ) : <EOL> self . _skip = count <EOL> return self <EOL> def limit ( self , count ) : <EOL> self . _limit = count if count != <NUM_LIT:0> else None <EOL> return self <EOL> def batch_size ( self , count ) : <EOL> return self <EOL> def close ( self ) : <EOL> pass <EOL> def distinct ( self , key ) : <EOL> if not isinstance ( key , helpers . basestring ) : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> unique = set ( ) <EOL> unique_dict_vals = [ ] <EOL> for x in iter ( self . _dataset ) : <EOL> value = _resolve_key ( key , x ) <EOL> if value == NOTHING : <EOL> continue <EOL> if isinstance ( value , dict ) : <EOL> if any ( dict_val == value for dict_val in unique_dict_vals ) : <EOL> continue <EOL> unique_dict_vals . append ( value ) <EOL> else : <EOL> unique . update ( <EOL> value if isinstance ( <EOL> value , ( tuple , list ) ) else [ value ] ) <EOL> return list ( unique ) + unique_dict_vals <EOL> def __getitem__ ( self , index ) : <EOL> arr = [ x for x in self . _dataset ] <EOL> self . _dataset = iter ( arr ) <EOL> return arr [ index ] <EOL> def _set_updater ( doc , field_name , value ) : <EOL> if isinstance ( value , ( tuple , list ) ) : <EOL> value = copy . deepcopy ( value ) <EOL> if isinstance ( doc , dict ) : <EOL> doc [ field_name ] = value <EOL> def _inc_updater ( doc , field_name , value ) : <EOL> if isinstance ( doc , dict ) : <EOL> doc [ field_name ] = doc . get ( field_name , <NUM_LIT:0> ) + value <EOL> def _sum_updater ( doc , field_name , current , result ) : <EOL> if isinstance ( doc , dict ) : <EOL> result = current + doc . get [ field_name , <NUM_LIT:0> ] <EOL> return result </s>
