# Source code for cassandra.cqlengine.query

    # Copyright DataStax, Inc.
    #
    # Licensed under the Apache License, Version 2.0 (the "License");
    # you may not use this file except in compliance with the License.
    # You may obtain a copy of the License at
    #
    # http://www.apache.org/licenses/LICENSE-2.0
    #
    # Unless required by applicable law or agreed to in writing, software
    # distributed under the License is distributed on an "AS IS" BASIS,
    # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    # See the License for the specific language governing permissions and
    # limitations under the License.

    import copy
    from datetime import datetime, timedelta
    from functools import partial
    import time
    from warnings import warn

    from cassandra.query import SimpleStatement, BatchType as CBatchType, BatchStatement
    from cassandra.cqlengine import columns, CQLEngineException, ValidationError, UnicodeMixin
    from cassandra.cqlengine import connection as conn
    from cassandra.cqlengine.functions import Token, BaseQueryFunction, QueryValue
    from cassandra.cqlengine.operators import (InOperator, EqualsOperator, GreaterThanOperator,
                                               GreaterThanOrEqualOperator, LessThanOperator,
                                               LessThanOrEqualOperator, ContainsOperator, BaseWhereOperator)
    from cassandra.cqlengine.statements import (WhereClause, SelectStatement, DeleteStatement,
                                                UpdateStatement, InsertStatement,
                                                BaseCQLStatement, MapDeleteClause, ConditionalClause)

    class QueryException(CQLEngineException):
        pass

    class IfNotExistsWithCounterColumn(CQLEngineException):
        pass

    class IfExistsWithCounterColumn(CQLEngineException):
        pass

    class LWTException(CQLEngineException):
        """Lightweight conditional exception.

        This exception will be raised when a write using an `IF` clause could not be
        applied due to existing data violating the condition. The existing data is
        available through the ``existing`` attribute.

        :param existing: The current state of the data which prevented the write.
        """
        def __init__(self, existing):
            super(LWTException, self).__init__("LWT Query was not applied")
            self.existing = existing

    class DoesNotExist(QueryException):
        pass

    class MultipleObjectsReturned(QueryException):
        pass

    def check_applied(result):
        """
        Raises LWTException if it looks like a failed LWT request. A LWTException
        won't be raised in the special case in which there are several failed LWT
        in a  :class:`~cqlengine.query.BatchQuery`.
        """
        try:
            applied = result.was_applied
        except Exception:
            applied = True  # result was not LWT form
        if not applied:
            raise LWTException(result.one())

    class AbstractQueryableColumn(UnicodeMixin):
        """
        exposes cql query operators through pythons
        builtin comparator symbols
        """

        def _get_column(self):
            raise NotImplementedError

        def __unicode__(self):
            raise NotImplementedError

        def _to_database(self, val):
            if isinstance(val, QueryValue):
                return val
            else:
                return self._get_column().to_database(val)

        def in_(self, item):
            """
            Returns an in operator

            used where you'd typically want to use python's `in` operator
            """
            return WhereClause(str(self), InOperator(), item)

        def contains_(self, item):
            """
            Returns a CONTAINS operator
            """
            return WhereClause(str(self), ContainsOperator(), item)

        def __eq__(self, other):
            return WhereClause(str(self), EqualsOperator(), self._to_database(other))

        def __gt__(self, other):
            return WhereClause(str(self), GreaterThanOperator(), self._to_database(other))

        def __ge__(self, other):
            return WhereClause(str(self), GreaterThanOrEqualOperator(), self._to_database(other))

        def __lt__(self, other):
            return WhereClause(str(self), LessThanOperator(), self._to_database(other))

        def __le__(self, other):
            return WhereClause(str(self), LessThanOrEqualOperator(), self._to_database(other))

    class BatchType(object):
        Unlogged = 'UNLOGGED'
        Counter = 'COUNTER'

    class BatchQuery(object):
        """
        Handles the batching of queries

        http://docs.datastax.com/en/cql/3.0/cql/cql_reference/batch_r.html

        See :doc:`/cqlengine/batches` for more details.
        """
        warn_multiple_exec = True

        _consistency = None

        _connection = None
        _connection_explicit = False

        def __init__(self, batch_type=None, timestamp=None, consistency=None, execute_on_exception=False,
                     timeout=conn.NOT_SET, connection=None):
            """
            :param batch_type: (optional) One of batch type values available through BatchType enum
            :type batch_type: BatchType, str or None
            :param timestamp: (optional) A datetime or timedelta object with desired timestamp to be applied
                to the batch conditional.
            :type timestamp: datetime or timedelta or None
            :param consistency: (optional) One of consistency values ("ANY", "ONE", "QUORUM" etc)
            :type consistency: The :class:`.ConsistencyLevel` to be used for the batch query, or None.
            :param execute_on_exception: (Defaults to False) Indicates that when the BatchQuery instance is used
                as a context manager the queries accumulated within the context must be executed despite
                encountering an error within the context. By default, any exception raised from within
                the context scope will cause the batched queries not to be executed.
            :type execute_on_exception: bool
            :param timeout: (optional) Timeout for the entire batch (in seconds), if not specified fallback
                to default session timeout
            :type timeout: float or None
            :param str connection: Connection name to use for the batch execution
            """
            self.queries = []
            self.batch_type = batch_type
            if timestamp is not None and not isinstance(timestamp, (datetime, timedelta)):
                raise CQLEngineException('timestamp object must be an instance of datetime')
            self.timestamp = timestamp
            self._consistency = consistency
            self._execute_on_exception = execute_on_exception
            self._timeout = timeout
            self._callbacks = []
            self._executed = False
            self._context_entered = False
            self._connection = connection
            if connection:
                self._connection_explicit = True

        def add_query(self, query):
            if not isinstance(query, BaseCQLStatement):
                raise CQLEngineException('only BaseCQLStatements can be added to a batch query')
            self.queries.append(query)

        def consistency(self, consistency):
            self._consistency = consistency

        def _execute_callbacks(self):
            for callback, args, kwargs in self._callbacks:
                callback(*args, **kwargs)

        def add_callback(self, fn, *args, **kwargs):
            """Add a function and arguments to be passed to it to be executed after the batch executes.

            A batch can support multiple callbacks.

            Note, that if the batch does not execute, the callbacks are not executed.
            A callback, thus, is an "on batch success" handler.

            :param fn: Callable object
            :type fn: callable
            :param \*args: Positional arguments to be passed to the callback at the time of execution
            :param \*\*kwargs: Named arguments to be passed to the callback at the time of execution
            """
            if not callable(fn):
                raise ValueError("Value for argument 'fn' is {0} and is not a callable object.".format(type(fn)))
            self._callbacks.append((fn, args, kwargs))

        def execute(self):
            if self._executed and self.warn_multiple_exec:
                msg = "Batch executed multiple times."
                if self._context_entered:
                    msg += " If using the batch as a context manager, there is no need to call execute directly."
                warn(msg)
            self._executed = True

            if len(self.queries) == 0:
                # Empty batch is a no-op
                # except for callbacks
                self._execute_callbacks()
                return

            batch_type = None if self.batch_type is CBatchType.LOGGED else self.batch_type
            opener = 'BEGIN ' + (str(batch_type) + ' ' if batch_type else '') + ' BATCH'
            if self.timestamp:

                if isinstance(self.timestamp, int):
                    ts = self.timestamp
                elif isinstance(self.timestamp, (datetime, timedelta)):
                    ts = self.timestamp
                    if isinstance(self.timestamp, timedelta):
                        ts += datetime.now()  # Apply timedelta
                    ts = int(time.mktime(ts.timetuple()) * 1e+6 + ts.microsecond)
                else:
                    raise ValueError("Batch expects a long, a timedelta, or a datetime")

                opener += ' USING TIMESTAMP {0}'.format(ts)

            query_list = [opener]
            parameters = {}
            ctx_counter = 0
            for query in self.queries:
                query.update_context_id(ctx_counter)
                ctx = query.get_context()
                ctx_counter += len(ctx)
                query_list.append('  ' + str(query))
                parameters.update(ctx)

            query_list.append('APPLY BATCH;')

            tmp = conn.execute('\n'.join(query_list), parameters, self._consistency, self._timeout, connection=self._connection)
            check_applied(tmp)

            self.queries = []
            self._execute_callbacks()

        def __enter__(self):
            self._context_entered = True
            return self

        def __exit__(self, exc_type, exc_val, exc_tb):
            # don't execute if there was an exception by default
            if exc_type is not None and not self._execute_on_exception:
                return
            self.execute()

    class ContextQuery(object):
        """
        A Context manager to allow a Model to switch context easily. Presently, the context only
        specifies a keyspace for model IO.

        :param \*args: One or more models. A model should be a class type, not an instance.
        :param \*\*kwargs: (optional) Context parameters: can be *keyspace* or *connection*

        For example:

        .. code-block:: python

                with ContextQuery(Automobile, keyspace='test2') as A:
                    A.objects.create(manufacturer='honda', year=2008, model='civic')
                    print(len(A.objects.all()))  # 1 result

                with ContextQuery(Automobile, keyspace='test4') as A:
                    print(len(A.objects.all()))  # 0 result

                # Multiple models
                with ContextQuery(Automobile, Automobile2, connection='cluster2') as (A, A2):
                    print(len(A.objects.all()))
                    print(len(A2.objects.all()))

        """

        def __init__(self, *args, **kwargs):
            from cassandra.cqlengine import models

            self.models = []

            if len(args) < 1:
                raise ValueError("No model provided.")

            keyspace = kwargs.pop('keyspace', None)
            connection = kwargs.pop('connection', None)

            if kwargs:
                raise ValueError("Unknown keyword argument(s): {0}".format(
                    ','.join(kwargs.keys())))

            for model in args:
                try:
                    issubclass(model, models.Model)
                except TypeError:
                    raise ValueError("Models must be derived from base Model.")

                m = models._clone_model_class(model, {})

                if keyspace:
                    m.__keyspace__ = keyspace
                if connection:
                    m.__connection__ = connection

                self.models.append(m)

        def __enter__(self):
            if len(self.models) > 1:
                return tuple(self.models)
            return self.models[0]

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

    class AbstractQuerySet(object):

        def __init__(self, model):
            super(AbstractQuerySet, self).__init__()
            self.model = model

            # Where clause filters
            self._where = []

            # Conditional clause filters
            self._conditional = []

            # ordering arguments
            self._order = []

            self._allow_filtering = False

            # CQL has a default limit of 10000, it's defined here
            # because explicit is better than implicit
            self._limit = 10000

            # We store the fields for which we use the Equal operator
            # in a query, so we don't select it from the DB. _defer_fields
            # will contain the names of the fields in the DB, not the names
            # of the variables used by the mapper
            self._defer_fields = set()
            self._deferred_values = {}

            # This variable will hold the names in the database of the fields
            # for which we want to query
            self._only_fields = []

            self._values_list = False
            self._flat_values_list = False

            # results cache
            self._result_cache = None
            self._result_idx = None
            self._result_generator = None
            self._materialize_results = True

            self._distinct_fields = None

            self._count = None

            self._batch = None
            self._ttl =  None
            self._consistency = None
            self._timestamp = None
            self._if_not_exists = False
            self._timeout = conn.NOT_SET
            self._if_exists = False
            self._fetch_size = None
            self._connection = None

        @property
        def column_family_name(self):
            return self.model.column_family_name()

        def _execute(self, statement):
            if self._batch:
                return self._batch.add_query(statement)
            else:
                connection = self._connection or self.model._get_connection()
                result = _execute_statement(self.model, statement, self._consistency, self._timeout, connection=connection)
                if self._if_not_exists or self._if_exists or self._conditional:
                    check_applied(result)
                return result

        def __unicode__(self):
            return str(self._select_query())

        def __str__(self):
            return str(self.__unicode__())

        def __call__(self, *args, **kwargs):
            return self.filter(*args, **kwargs)

        def __deepcopy__(self, memo):
            clone = self.__class__(self.model)
            for k, v in self.__dict__.items():
                if k in ['_con', '_cur', '_result_cache', '_result_idx', '_result_generator', '_construct_result']:  # don't clone these, which are per-request-execution
                    clone.__dict__[k] = None
                elif k == '_batch':
                    # we need to keep the same batch instance across
                    # all queryset clones, otherwise the batched queries
                    # fly off into other batch instances which are never
                    # executed, thx @dokai
                    clone.__dict__[k] = self._batch
                elif k == '_timeout':
                    clone.__dict__[k] = self._timeout
                else:
                    clone.__dict__[k] = copy.deepcopy(v, memo)

            return clone

        def __len__(self):
            self._execute_query()
            return self.count()

        # ----query generation / execution----

        def _select_fields(self):
            """ returns the fields to select """
            return []

        def _validate_select_where(self):
            """ put select query validation here """

        def _select_query(self):
            """
            Returns a select clause based on the given filter args
            """
            if self._where:
                self._validate_select_where()
            return SelectStatement(
                self.column_family_name,
                fields=self._select_fields(),
                where=self._where,
                order_by=self._order,
                limit=self._limit,
                allow_filtering=self._allow_filtering,
                distinct_fields=self._distinct_fields,
                fetch_size=self._fetch_size
            )

        # ----Reads------

        def _execute_query(self):
            if self._batch:
                raise CQLEngineException("Only inserts, updates, and deletes are available in batch mode")
            if self._result_cache is None:
                self._result_generator = (i for i in self._execute(self._select_query()))
                self._result_cache = []
                self._construct_result = self._maybe_inject_deferred(self._get_result_constructor())

                # "DISTINCT COUNT()" is not supported in C* < 2.2, so we need to materialize all results to get
                # len() and count() working with DISTINCT queries
                if self._materialize_results or self._distinct_fields:
                    self._fill_result_cache()

        def _fill_result_cache(self):
            """
            Fill the result cache with all results.
            """

            idx = 0
            try:
                while True:
                    idx += 1000
                    self._fill_result_cache_to_idx(idx)
            except StopIteration:
                pass

            self._count = len(self._result_cache)

        def _fill_result_cache_to_idx(self, idx):
            self._execute_query()
            if self._result_idx is None:
                self._result_idx = -1

            qty = idx - self._result_idx
            if qty < 1:
                return
            else:
                for idx in range(qty):
                    self._result_idx += 1
                    while True:
                        try:
                            self._result_cache[self._result_idx] = self._construct_result(self._result_cache[self._result_idx])
                            break
                        except IndexError:
                            self._result_cache.append(next(self._result_generator))

        def __iter__(self):
            self._execute_query()

            idx = 0
            while True:
                if len(self._result_cache) <= idx:
                    try:
                        self._result_cache.append(next(self._result_generator))
                    except StopIteration:
                        break

                instance = self._result_cache[idx]
                if isinstance(instance, dict):
                    self._fill_result_cache_to_idx(idx)
                yield self._result_cache[idx]

                idx += 1

        def __getitem__(self, s):
            self._execute_query()

            if isinstance(s, slice):
                start = s.start if s.start else 0

                if start < 0 or (s.stop is not None and s.stop < 0):
                    warn("ModelQuerySet slicing with negative indices support will be removed in 4.0.",
                         DeprecationWarning)

                # calculate the amount of results that need to be loaded
                end = s.stop
                if start < 0 or s.stop is None or s.stop < 0:
                    end = self.count()

                try:
                    self._fill_result_cache_to_idx(end)
                except StopIteration:
                    pass

                return self._result_cache[start:s.stop:s.step]
            else:
                try:
                    s = int(s)
                except (ValueError, TypeError):
                    raise TypeError('QuerySet indices must be integers')

                if s < 0:
                    warn("ModelQuerySet indexing with negative indices support will be removed in 4.0.",
                         DeprecationWarning)

                # Using negative indexing is costly since we have to execute a count()
                if s < 0:
                    num_results = self.count()
                    s += num_results

                try:
                    self._fill_result_cache_to_idx(s)
                except StopIteration:
                    raise IndexError

                return self._result_cache[s]

        def _get_result_constructor(self):
            """
            Returns a function that will be used to instantiate query results
            """
            raise NotImplementedError

        @staticmethod
        def _construct_with_deferred(f, deferred, row):
            row.update(deferred)
            return f(row)

        def _maybe_inject_deferred(self, constructor):
            return partial(self._construct_with_deferred, constructor, self._deferred_values)\
                if self._deferred_values else constructor

        def batch(self, batch_obj):
            """
            Set a batch object to run the query on.

            Note: running a select query with a batch object will raise an exception
            """
            if self._connection:
                raise CQLEngineException("Cannot specify the connection on model in batch mode.")

            if batch_obj is not None and not isinstance(batch_obj, BatchQuery):
                raise CQLEngineException('batch_obj must be a BatchQuery instance or None')
            clone = copy.deepcopy(self)
            clone._batch = batch_obj
            return clone

        def first(self):
            try:
                return next(iter(self))
            except StopIteration:
                return None

        def all(self):
            """
            Returns a queryset matching all rows

            .. code-block:: python

                for user in User.objects().all():
                    print(user)
            """
            return copy.deepcopy(self)

        def consistency(self, consistency):
            """
            Sets the consistency level for the operation. See :class:`.ConsistencyLevel`.

            .. code-block:: python

                for user in User.objects(id=3).consistency(CL.ONE):
                    print(user)
            """
            clone = copy.deepcopy(self)
            clone._consistency = consistency
            return clone

        def _parse_filter_arg(self, arg):
            """
            Parses a filter arg in the format:
            <colname>__<op>
            :returns: colname, op tuple
            """
            statement = arg.rsplit('__', 1)
            if len(statement) == 1:
                return arg, None
            elif len(statement) == 2:
                return (statement[0], statement[1]) if arg != 'pk__token' else (arg, None)
            else:
                raise QueryException("Can't parse '{0}'".format(arg))

        def iff(self, *args, **kwargs):
            """Adds IF statements to queryset"""
            if len([x for x in kwargs.values() if x is None]):
                raise CQLEngineException("None values on iff are not allowed")

            clone = copy.deepcopy(self)
            for operator in args:
                if not isinstance(operator, ConditionalClause):
                    raise QueryException('{0} is not a valid query operator'.format(operator))
                clone._conditional.append(operator)

            for arg, val in kwargs.items():
                if isinstance(val, Token):
                    raise QueryException("Token() values are not valid in conditionals")

                col_name, col_op = self._parse_filter_arg(arg)
                try:
                    column = self.model._get_column(col_name)
                except KeyError:
                    raise QueryException("Can't resolve column name: '{0}'".format(col_name))

                if isinstance(val, BaseQueryFunction):
                    query_val = val
                else:
                    query_val = column.to_database(val)

                operator_class = BaseWhereOperator.get_operator(col_op or 'EQ')
                operator = operator_class()
                clone._conditional.append(WhereClause(column.db_field_name, operator, query_val))

            return clone

        def filter(self, *args, **kwargs):
            """
            Adds WHERE arguments to the queryset, returning a new queryset

            See :ref:`retrieving-objects-with-filters`

            Returns a QuerySet filtered on the keyword arguments
            """
            # add arguments to the where clause filters
            if len([x for x in kwargs.values() if x is None]):
                raise CQLEngineException("None values on filter are not allowed")

            clone = copy.deepcopy(self)
            for operator in args:
                if not isinstance(operator, WhereClause):
                    raise QueryException('{0} is not a valid query operator'.format(operator))
                clone._where.append(operator)

            for arg, val in kwargs.items():
                col_name, col_op = self._parse_filter_arg(arg)
                quote_field = True

                if not isinstance(val, Token):
                    try:
                        column = self.model._get_column(col_name)
                    except KeyError:
                        raise QueryException("Can't resolve column name: '{0}'".format(col_name))
                else:
                    if col_name != 'pk__token':
                        raise QueryException("Token() values may only be compared to the 'pk__token' virtual column")

                    column = columns._PartitionKeysToken(self.model)
                    quote_field = False

                    partition_columns = column.partition_columns
                    if len(partition_columns) != len(val.value):
                        raise QueryException(
                            'Token() received {0} arguments but model has {1} partition keys'.format(
                                len(val.value), len(partition_columns)))
                    val.set_columns(partition_columns)

                # get query operator, or use equals if not supplied
                operator_class = BaseWhereOperator.get_operator(col_op or 'EQ')
                operator = operator_class()

                if isinstance(operator, InOperator):
                    if not isinstance(val, (list, tuple)):
                        raise QueryException('IN queries must use a list/tuple value')
                    query_val = [column.to_database(v) for v in val]
                elif isinstance(val, BaseQueryFunction):
                    query_val = val
                elif (isinstance(operator, ContainsOperator) and
                      isinstance(column, (columns.List, columns.Set, columns.Map))):
                    # For ContainsOperator and collections, we query using the value, not the container
                    query_val = val
                else:
                    query_val = column.to_database(val)
                    if not col_op:  # only equal values should be deferred
                        clone._defer_fields.add(column.db_field_name)
                        clone._deferred_values[column.db_field_name] = val  # map by db field name for substitution in results

                clone._where.append(WhereClause(column.db_field_name, operator, query_val, quote_field=quote_field))

            return clone

        def get(self, *args, **kwargs):
            """
            Returns a single instance matching this query, optionally with additional filter kwargs.

            See :ref:`retrieving-objects-with-filters`

            Returns a single object matching the QuerySet.

            .. code-block:: python

                user = User.get(id=1)

            If no objects are matched, a :class:`~.DoesNotExist` exception is raised.

            If more than one object is found, a :class:`~.MultipleObjectsReturned` exception is raised.
            """
            if args or kwargs:
                return self.filter(*args, **kwargs).get()

            self._execute_query()

            # Check that the resultset only contains one element, avoiding sending a COUNT query
            try:
                self[1]
                raise self.model.MultipleObjectsReturned('Multiple objects found')
            except IndexError:
                pass

            try:
                obj = self[0]
            except IndexError:
                raise self.model.DoesNotExist

            return obj

        def _get_ordering_condition(self, colname):
            order_type = 'DESC' if colname.startswith('-') else 'ASC'
            colname = colname.replace('-', '')

            return colname, order_type

        def order_by(self, *colnames):
            """
            Sets the column(s) to be used for ordering

            Default order is ascending, prepend a '-' to any column name for descending

            *Note: column names must be a clustering key*

            .. code-block:: python

                from uuid import uuid1,uuid4

                class Comment(Model):
                    photo_id = UUID(primary_key=True)
                    comment_id = TimeUUID(primary_key=True, default=uuid1) # second primary key component is a clustering key
                    comment = Text()

                sync_table(Comment)

                u = uuid4()
                for x in range(5):
                    Comment.create(photo_id=u, comment="test %d" % x)

                print("Normal")
                for comment in Comment.objects(photo_id=u):
                    print(comment.comment_id)

                print("Reversed")
                for comment in Comment.objects(photo_id=u).order_by("-comment_id"):
                    print(comment.comment_id)
            """
            if len(colnames) == 0:
                clone = copy.deepcopy(self)
                clone._order = []
                return clone

            conditions = []
            for colname in colnames:
                conditions.append('"{0}" {1}'.format(*self._get_ordering_condition(colname)))

            clone = copy.deepcopy(self)
            clone._order.extend(conditions)
            return clone

        def count(self):
            """
            Returns the number of rows matched by this query.

            *Note: This function executes a SELECT COUNT() and has a performance cost on large datasets*
            """
            if self._batch:
                raise CQLEngineException("Only inserts, updates, and deletes are available in batch mode")

            if self._count is None:
                query = self._select_query()
                query.count = True
                result = self._execute(query)
                count_row = result.one().popitem()
                self._count = count_row[1]
            return self._count

        def distinct(self, distinct_fields=None):
            """
            Returns the DISTINCT rows matched by this query.

            distinct_fields default to the partition key fields if not specified.

            *Note: distinct_fields must be a partition key or a static column*

            .. code-block:: python

                class Automobile(Model):
                    manufacturer = columns.Text(partition_key=True)
                    year = columns.Integer(primary_key=True)
                    model = columns.Text(primary_key=True)
                    price = columns.Decimal()

                sync_table(Automobile)

                # create rows

                Automobile.objects.distinct()

                # or

                Automobile.objects.distinct(['manufacturer'])

            """

            clone = copy.deepcopy(self)
            if distinct_fields:
                clone._distinct_fields = distinct_fields
            else:
                clone._distinct_fields = [x.column_name for x in self.model._partition_keys.values()]

            return clone

        def limit(self, v):
            """
            Limits the number of results returned by Cassandra. Use *0* or *None* to disable.

            *Note that CQL's default limit is 10,000, so all queries without a limit set explicitly will have an implicit limit of 10,000*

            .. code-block:: python

                # Fetch 100 users
                for user in User.objects().limit(100):
                    print(user)

                # Fetch all users
                for user in User.objects().limit(None):
                    print(user)
            """

            if v is None:
                v = 0

            if not isinstance(v, int):
                raise TypeError
            if v == self._limit:
                return self

            if v < 0:
                raise QueryException("Negative limit is not allowed")

            clone = copy.deepcopy(self)
            clone._limit = v
            return clone

        def fetch_size(self, v):
            """
            Sets the number of rows that are fetched at a time.

            *Note that driver's default fetch size is 5000.*

            .. code-block:: python

                for user in User.objects().fetch_size(500):
                    print(user)
            """

            if not isinstance(v, int):
                raise TypeError
            if v == self._fetch_size:
                return self

            if v < 1:
                raise QueryException("fetch size less than 1 is not allowed")

            clone = copy.deepcopy(self)
            clone._fetch_size = v
            return clone

        def allow_filtering(self):
            """
            Enables the (usually) unwise practive of querying on a clustering key without also defining a partition key
            """
            clone = copy.deepcopy(self)
            clone._allow_filtering = True
            return clone

        def _only_or_defer(self, action, fields):
            if action == 'only' and self._only_fields:
                raise QueryException("QuerySet already has 'only' fields defined")

            clone = copy.deepcopy(self)

            # check for strange fields
            missing_fields = [f for f in fields if f not in self.model._columns.keys()]
            if missing_fields:
                raise QueryException(
                    "Can't resolve fields {0} in {1}".format(
                        ', '.join(missing_fields), self.model.__name__))

            fields = [self.model._columns[field].db_field_name for field in fields]

            if action == 'defer':
                clone._defer_fields.update(fields)
            elif action == 'only':
                clone._only_fields = fields
            else:
                raise ValueError

            return clone

        def only(self, fields):
            """ Load only these fields for the returned query """
            return self._only_or_defer('only', fields)

        def defer(self, fields):
            """ Don't load these fields for the returned query """
            return self._only_or_defer('defer', fields)

        def create(self, **kwargs):
            return self.model(**kwargs) \
                .batch(self._batch) \
                .ttl(self._ttl) \
                .consistency(self._consistency) \
                .if_not_exists(self._if_not_exists) \
                .timestamp(self._timestamp) \
                .if_exists(self._if_exists) \
                .using(connection=self._connection) \
                .save()

        def delete(self):
            """
            Deletes the contents of a query
            """
            # validate where clause
            partition_keys = set(x.db_field_name for x in self.model._partition_keys.values())
            if partition_keys - set(c.field for c in self._where):
                raise QueryException("The partition key must be defined on delete queries")

            dq = DeleteStatement(
                self.column_family_name,
                where=self._where,
                timestamp=self._timestamp,
                conditionals=self._conditional,
                if_exists=self._if_exists
            )
            self._execute(dq)

        def __eq__(self, q):
            if len(self._where) == len(q._where):
                return all([w in q._where for w in self._where])
            return False

        def __ne__(self, q):
            return not (self != q)

        def timeout(self, timeout):
            """
            :param timeout: Timeout for the query (in seconds)
            :type timeout: float or None
            """
            clone = copy.deepcopy(self)
            clone._timeout = timeout
            return clone

        def using(self, keyspace=None, connection=None):
            """
            Change the context on-the-fly of the Model class (keyspace, connection)
            """

            if connection and self._batch:
                raise CQLEngineException("Cannot specify a connection on model in batch mode.")

            clone = copy.deepcopy(self)
            if keyspace:
                from cassandra.cqlengine.models import _clone_model_class
                clone.model = _clone_model_class(self.model, {'__keyspace__': keyspace})

            if connection:
                clone._connection = connection

            return clone

    class ResultObject(dict):
        """
        adds attribute access to a dictionary
        """

        def __getattr__(self, item):
            try:
                return self[item]
            except KeyError:
                raise AttributeError

    class SimpleQuerySet(AbstractQuerySet):
        """
        Overrides _get_result_constructor for querysets that do not define a model (e.g. NamedTable queries)
        """

        def _get_result_constructor(self):
            """
            Returns a function that will be used to instantiate query results
            """
            return ResultObject

    class ModelQuerySet(AbstractQuerySet):
        """
        """
        def _validate_select_where(self):
            """ Checks that a filterset will not create invalid select statement """
            # check that there's either a =, a IN or a CONTAINS (collection)
            # relationship with a primary key or indexed field. We also allow
            # custom indexes to be queried with any operator (a difference
            # between a secondary index)
            equal_ops = [self.model._get_column_by_db_name(w.field) \
                         for w in self._where if not isinstance(w.value, Token)
                         and (isinstance(w.operator, EqualsOperator)
                              or self.model._get_column_by_db_name(w.field).custom_index)]
            token_comparison = any([w for w in self._where if isinstance(w.value, Token)])
            if not any(w.primary_key or w.has_index for w in equal_ops) and not token_comparison and not self._allow_filtering:
                raise QueryException(
                    ('Where clauses require either  =, a IN or a CONTAINS '
                     '(collection) comparison with either a primary key or '
                     'indexed field. You might want to consider setting '
                     'custom_index on fields that you manage index outside '
                     'cqlengine.'))

            if not self._allow_filtering:
                # if the query is not on an indexed field
                if not any(w.has_index for w in equal_ops):
                    if not any([w.partition_key for w in equal_ops]) and not token_comparison:
                        raise QueryException(
                            ('Filtering on a clustering key without a partition '
                             'key is not allowed unless allow_filtering() is '
                             'called on the queryset. You might want to consider '
                             'setting custom_index on fields that you manage '
                             'index outside cqlengine.'))

        def _select_fields(self):
            if self._defer_fields or self._only_fields:
                fields = [columns.db_field_name for columns in self.model._columns.values()]
                if self._defer_fields:
                    fields = [f for f in fields if f not in self._defer_fields]
                    # select the partition keys if all model fields are set defer
                    if not fields:
                        fields = [columns.db_field_name for columns in self.model._partition_keys.values()]
                if self._only_fields:
                    fields = [f for f in fields if f in self._only_fields]
                if not fields:
                    raise QueryException('No fields in select query. Only fields: "{0}", defer fields: "{1}"'.format(
                        ','.join(self._only_fields), ','.join(self._defer_fields)))
                return fields
            return super(ModelQuerySet, self)._select_fields()

        def _get_result_constructor(self):
            """ Returns a function that will be used to instantiate query results """
            if not self._values_list:  # we want models
                return self.model._construct_instance
            elif self._flat_values_list:  # the user has requested flattened list (1 value per row)
                key = self._only_fields[0]
                return lambda row: row[key]
            else:
                return lambda row: [row[f] for f in self._only_fields]

        def _get_ordering_condition(self, colname):
            colname, order_type = super(ModelQuerySet, self)._get_ordering_condition(colname)

            column = self.model._columns.get(colname)
            if column is None:
                raise QueryException("Can't resolve the column name: '{0}'".format(colname))

            # validate the column selection
            if not column.primary_key:
                raise QueryException(
                    "Can't order on '{0}', can only order on (clustered) primary keys".format(colname))

            pks = [v for k, v in self.model._columns.items() if v.primary_key]
            if column == pks[0]:
                raise QueryException(
                    "Can't order by the first primary key (partition key), clustering (secondary) keys only")

            return column.db_field_name, order_type

        def values_list(self, *fields, **kwargs):
            """ Instructs the query set to return tuples, not model instance """
            flat = kwargs.pop('flat', False)
            if kwargs:
                raise TypeError('Unexpected keyword arguments to values_list: %s'
                                % (kwargs.keys(),))
            if flat and len(fields) > 1:
                raise TypeError("'flat' is not valid when values_list is called with more than one field.")
            clone = self.only(fields)
            clone._values_list = True
            clone._flat_values_list = flat
            return clone

        def ttl(self, ttl):
            """
            Sets the ttl (in seconds) for modified data.

            *Note that running a select query with a ttl value will raise an exception*
            """
            clone = copy.deepcopy(self)
            clone._ttl = ttl
            return clone

        def timestamp(self, timestamp):
            """
            Allows for custom timestamps to be saved with the record.
            """
            clone = copy.deepcopy(self)
            clone._timestamp = timestamp
            return clone

        def if_not_exists(self):
            """
            Check the existence of an object before insertion.

            If the insertion isn't applied, a LWTException is raised.
            """
            if self.model._has_counter:
                raise IfNotExistsWithCounterColumn('if_not_exists cannot be used with tables containing counter columns')
            clone = copy.deepcopy(self)
            clone._if_not_exists = True
            return clone

        def if_exists(self):
            """
            Check the existence of an object before an update or delete.

            If the update or delete isn't applied, a LWTException is raised.
            """
            if self.model._has_counter:
                raise IfExistsWithCounterColumn('if_exists cannot be used with tables containing counter columns')
            clone = copy.deepcopy(self)
            clone._if_exists = True
            return clone

        def update(self, **values):
            """
            Performs an update on the row selected by the queryset. Include values to update in the
            update like so:

            .. code-block:: python

                Model.objects(key=n).update(value='x')

            Passing in updates for columns which are not part of the model will raise a ValidationError.

            Per column validation will be performed, but instance level validation will not
            (i.e., `Model.validate` is not called).  This is sometimes referred to as a blind update.

            For example:

            .. code-block:: python

                class User(Model):
                    id = Integer(primary_key=True)
                    name = Text()

                setup(["localhost"], "test")
                sync_table(User)

                u = User.create(id=1, name="jon")

                User.objects(id=1).update(name="Steve")

                # sets name to null
                User.objects(id=1).update(name=None)

            Also supported is blindly adding and removing elements from container columns,
            without loading a model instance from Cassandra.

            Using the syntax `.update(column_name={x, y, z})` will overwrite the contents of the container, like updating a
            non container column. However, adding `__<operation>` to the end of the keyword arg, makes the update call add
            or remove items from the collection, without overwriting then entire column.

            Given the model below, here are the operations that can be performed on the different container columns:

            .. code-block:: python

                class Row(Model):
                    row_id      = columns.Integer(primary_key=True)
                    set_column  = columns.Set(Integer)
                    list_column = columns.List(Integer)
                    map_column  = columns.Map(Integer, Integer)

            :class:`~cqlengine.columns.Set`

            - `add`: adds the elements of the given set to the column
            - `remove`: removes the elements of the given set to the column

            .. code-block:: python

                # add elements to a set
                Row.objects(row_id=5).update(set_column__add={6})

                # remove elements to a set
                Row.objects(row_id=5).update(set_column__remove={4})

            :class:`~cqlengine.columns.List`

            - `append`: appends the elements of the given list to the end of the column
            - `prepend`: prepends the elements of the given list to the beginning of the column

            .. code-block:: python

                # append items to a list
                Row.objects(row_id=5).update(list_column__append=[6, 7])

                # prepend items to a list
                Row.objects(row_id=5).update(list_column__prepend=[1, 2])

            :class:`~cqlengine.columns.Map`

            - `update`: adds the given keys/values to the columns, creating new entries if they didn't exist, and overwriting old ones if they did

            .. code-block:: python

                # add items to a map
                Row.objects(row_id=5).update(map_column__update={1: 2, 3: 4})

                # remove items from a map
                Row.objects(row_id=5).update(map_column__remove={1, 2})
            """
            if not values:
                return

            nulled_columns = set()
            updated_columns = set()
            us = UpdateStatement(self.column_family_name, where=self._where, ttl=self._ttl,
                                 timestamp=self._timestamp, conditionals=self._conditional, if_exists=self._if_exists)
            for name, val in values.items():
                col_name, col_op = self._parse_filter_arg(name)
                col = self.model._columns.get(col_name)
                # check for nonexistant columns
                if col is None:
                    raise ValidationError("{0}.{1} has no column named: {2}".format(self.__module__, self.model.__name__, col_name))
                # check for primary key update attempts
                if col.is_primary_key:
                    raise ValidationError("Cannot apply update to primary key '{0}' for {1}.{2}".format(col_name, self.__module__, self.model.__name__))

                if col_op == 'remove' and isinstance(col, columns.Map):
                    if not isinstance(val, set):
                        raise ValidationError(
                            "Cannot apply update operation '{0}' on column '{1}' with value '{2}'. A set is required.".format(col_op, col_name, val))
                    val = {v: None for v in val}
                else:
                    # we should not provide default values in this use case.
                    val = col.validate(val)

                if val is None:
                    nulled_columns.add(col_name)
                    continue

                us.add_update(col, val, operation=col_op)
                updated_columns.add(col_name)

            if us.assignments:
                self._execute(us)

            if nulled_columns:
                delete_conditional = [condition for condition in self._conditional
                                      if condition.field not in updated_columns] if self._conditional else None
                ds = DeleteStatement(self.column_family_name, fields=nulled_columns,
                                     where=self._where, conditionals=delete_conditional, if_exists=self._if_exists)
                self._execute(ds)

    class DMLQuery(object):
        """
        A query object used for queries performing inserts, updates, or deletes

        this is usually instantiated by the model instance to be modified

        unlike the read query object, this is mutable
        """
        _ttl = None
        _consistency = None
        _timestamp = None
        _if_not_exists = False
        _if_exists = False

        def __init__(self, model, instance=None, batch=None, ttl=None, consistency=None, timestamp=None,
                     if_not_exists=False, conditional=None, timeout=conn.NOT_SET, if_exists=False):
            self.model = model
            self.column_family_name = self.model.column_family_name()
            self.instance = instance
            self._batch = batch
            self._ttl = ttl
            self._consistency = consistency
            self._timestamp = timestamp
            self._if_not_exists = if_not_exists
            self._if_exists = if_exists
            self._conditional = conditional
            self._timeout = timeout

        def _execute(self, statement):
            connection = self.instance._get_connection() if self.instance else self.model._get_connection()
            if self._batch:
                if self._batch._connection:
                    if not self._batch._connection_explicit and connection and \
                            connection != self._batch._connection:
                                raise CQLEngineException('BatchQuery queries must be executed on the same connection')
                else:
                    # set the BatchQuery connection from the model
                    self._batch._connection = connection
                return self._batch.add_query(statement)
            else:
                results = _execute_statement(self.model, statement, self._consistency, self._timeout, connection=connection)
                if self._if_not_exists or self._if_exists or self._conditional:
                    check_applied(results)
                return results

        def batch(self, batch_obj):
            if batch_obj is not None and not isinstance(batch_obj, BatchQuery):
                raise CQLEngineException('batch_obj must be a BatchQuery instance or None')
            self._batch = batch_obj
            return self

        def _delete_null_columns(self, conditionals=None):
            """
            executes a delete query to remove columns that have changed to null
            """
            ds = DeleteStatement(self.column_family_name, conditionals=conditionals, if_exists=self._if_exists)
            deleted_fields = False
            static_only = True
            for _, v in self.instance._values.items():
                col = v.column
                if v.deleted:
                    ds.add_field(col.db_field_name)
                    deleted_fields = True
                    static_only &= col.static
                elif isinstance(col, columns.Map):
                    uc = MapDeleteClause(col.db_field_name, v.value, v.previous_value)
                    if uc.get_context_size() > 0:
                        ds.add_field(uc)
                        deleted_fields = True
                        static_only |= col.static

            if deleted_fields:
                keys = self.model._partition_keys if static_only else self.model._primary_keys
                for name, col in keys.items():
                    ds.add_where(col, EqualsOperator(), getattr(self.instance, name))
                self._execute(ds)

        def update(self):
            """
            updates a row.
            This is a blind update call.
            All validation and cleaning needs to happen
            prior to calling this.
            """
            if self.instance is None:
                raise CQLEngineException("DML Query intance attribute is None")
            assert type(self.instance) == self.model
            null_clustering_key = False if len(self.instance._clustering_keys) == 0 else True
            static_changed_only = True
            statement = UpdateStatement(self.column_family_name, ttl=self._ttl, timestamp=self._timestamp,
                                        conditionals=self._conditional, if_exists=self._if_exists)
            for name, col in self.instance._clustering_keys.items():
                null_clustering_key = null_clustering_key and col._val_is_null(getattr(self.instance, name, None))

            updated_columns = set()
            # get defined fields and their column names
            for name, col in self.model._columns.items():
                # if clustering key is null, don't include non static columns
                if null_clustering_key and not col.static and not col.partition_key:
                    continue
                if not col.is_primary_key:
                    val = getattr(self.instance, name, None)
                    val_mgr = self.instance._values[name]

                    if val is None:
                        continue

                    if not val_mgr.changed and not isinstance(col, columns.Counter):
                        continue

                    static_changed_only = static_changed_only and col.static
                    statement.add_update(col, val, previous=val_mgr.previous_value)
                    updated_columns.add(col.db_field_name)

            if statement.assignments:
                for name, col in self.model._primary_keys.items():
                    # only include clustering key if clustering key is not null, and non static columns are changed to avoid cql error
                    if (null_clustering_key or static_changed_only) and (not col.partition_key):
                        continue
                    statement.add_where(col, EqualsOperator(), getattr(self.instance, name))
                self._execute(statement)

            if not null_clustering_key:
                # remove conditions on fields that have been updated
                delete_conditionals = [condition for condition in self._conditional
                                       if condition.field not in updated_columns] if self._conditional else None
                self._delete_null_columns(delete_conditionals)

        def save(self):
            """
            Creates / updates a row.
            This is a blind insert call.
            All validation and cleaning needs to happen
            prior to calling this.
            """
            if self.instance is None:
                raise CQLEngineException("DML Query intance attribute is None")
            assert type(self.instance) == self.model

            nulled_fields = set()
            if self.instance._has_counter or self.instance._can_update():
                if self.instance._has_counter:
                    warn("'create' and 'save' actions on Counters are deprecated. It will be disallowed in 4.0. "
                        "Use the 'update' mechanism instead.", DeprecationWarning)
                return self.update()
            else:
                insert = InsertStatement(self.column_family_name, ttl=self._ttl, timestamp=self._timestamp, if_not_exists=self._if_not_exists)
                static_save_only = False if len(self.instance._clustering_keys) == 0 else True
                for name, col in self.instance._clustering_keys.items():
                    static_save_only = static_save_only and col._val_is_null(getattr(self.instance, name, None))
                for name, col in self.instance._columns.items():
                    if static_save_only and not col.static and not col.partition_key:
                        continue
                    val = getattr(self.instance, name, None)
                    if col._val_is_null(val):
                        if self.instance._values[name].changed:
                            nulled_fields.add(col.db_field_name)
                        continue
                    if col.has_default and not self.instance._values[name].changed:
                        # Ensure default columns included in a save() are marked as explicit, to get them *persisted* properly
                        self.instance._values[name].explicit = True
                    insert.add_assignment(col, getattr(self.instance, name, None))

            # skip query execution if it's empty
            # caused by pointless update queries
            if not insert.is_empty:
                self._execute(insert)
            # delete any nulled columns
            if not static_save_only:
                self._delete_null_columns()

        def delete(self):
            """ Deletes one instance """
            if self.instance is None:
                raise CQLEngineException("DML Query instance attribute is None")

            ds = DeleteStatement(self.column_family_name, timestamp=self._timestamp, conditionals=self._conditional, if_exists=self._if_exists)
            for name, col in self.model._primary_keys.items():
                val = getattr(self.instance, name)
                if val is None and not col.partition_key:
                    continue
                ds.add_where(col, EqualsOperator(), val)
            self._execute(ds)

    def _execute_statement(model, statement, consistency_level, timeout, connection=None):
        params = statement.get_context()
        s = SimpleStatement(str(statement), consistency_level=consistency_level, fetch_size=statement.fetch_size)
        if model._partition_key_index:
            key_values = statement.partition_key_values(model._partition_key_index)
            if not any(v is None for v in key_values):
                parts = model._routing_key_from_values(key_values, conn.get_cluster(connection).protocol_version)
                s.routing_key = parts
                s.keyspace = model._get_keyspace()
        connection = connection or model._get_connection()
        return conn.execute(s, params, timeout=timeout, connection=connection)