<commit_before># Copyright 2010-2014 RethinkDB, all rights reserved.

import errno
import json
import numbers
import socket
import struct
import time
try:
    from importlib import import_module
except ImportError:
    def import_module(name, package=None):
        # song & dance needed to do relative import in 2.6, which
        # doesn't have importlib
        return __import__(name[1:], globals(), locals(), [], 1)

from . import ql2_pb2 as p

__all__ = ['connect', 'set_loop_type', 'Connection', 'Cursor']

pResponse = p.Response.ResponseType
pQuery = p.Query.QueryType

from . import repl              # For the repl connection
from .errors import *
from .ast import RqlQuery, RqlTopLevelQuery, DB
from .ast import recursively_convert_pseudotypes

try:
    xrange
except NameError:
    xrange = range

try:
    {}.iteritems
    dict_items = lambda d: d.iteritems()
except AttributeError:
    dict_items = lambda d: d.items()

def decodeUTF(inputPipe):
    # attempt to decode input as utf-8 with fallbacks to get something useful
    try:
        return inputPipe.decode('utf-8', errors='ignore')
    except TypeError:
        try:
            return inputPipe.decode('utf-8')
        except UnicodeError:
            return repr(inputPipe)

def convert_pseudo(value, query):
    return recursively_convert_pseudotypes(value, query.global_optargs)

def maybe_profile(value, res):
    if res.profile is not None:
        return {'value': value, 'profile': res.profile}
    return value

class Query(object):
    def __init__(self, type, token, term, global_optargs):
        self.type = type
        self.token = token
        self.term = term
        self.global_optargs = global_optargs

    def serialize(self):
        message = [self.type]
        if self.term is not None:
            message.append(self.term.build())
        if self.global_optargs is not None:
            optargs = {}
            for k, v in dict_items(self.global_optargs):
                optargs[k] = v.build() if isinstance(v, RqlQuery) else v
            message.append(optargs)
        query_str = json.dumps(message, ensure_ascii=False, allow_nan=False).encode('utf-8')
        query_header = struct.pack('<QL', self.token, len(query_str))
        return query_header + query_str


class Response(object):
    def __init__(self, token, json_str):
        try:
            json_str = json_str.decode('utf-8')
        except AttributeError:
            pass               # Python3 str objects are already utf-8
        self.token = token
        full_response = json.loads(json_str)
        self.type = full_response["t"]
        self.data = full_response["r"]
        self.backtrace = full_response.get("b", None)
        self.profile = full_response.get("p", None)

    def make_error(self, query):
        if self.type == pResponse.CLIENT_ERROR:
            return RqlClientError(self.data[0], query.term, self.backtrace)
        elif self.type == pResponse.COMPILE_ERROR:
            return RqlCompileError(self.data[0], query.term, self.backtrace)
        elif self.type == pResponse.RUNTIME_ERROR:
            return RqlRuntimeError(self.data[0], query.term, self.backtrace)
        return RqlDriverError("Unknown Response type %d encountered" +
                              " in a response." % self.type)


# This class encapsulates all shared behavior between cursor implementations.
# It provides iteration over the cursor using `iter`, as well as incremental
# iteration using `next`.
#
# query - the original query that resulted in the cursor, used for:
#     query.term - the term to be used for pretty-printing backtraces
#     query.token - the token to use for subsequent CONTINUE and STOP requests
#     query.global_optargs - dictate how to format results
# items - The current list of items obtained from the server, this is
#     added to in `_extend`, which is called by the ConnectionInstance when a
#     new response arrives for this cursor.
# outstanding_requests - The number of requests that are currently awaiting
#     a response from the server.  This will typically be 0 or 1 unless the
#     cursor is exhausted, but this can be higher if `close` is called.
# threshold - a CONTINUE request will be sent when the length of `items` goes
#     below this number.
# error - indicates the current state of the cursor:
#     None - there is more data available from the server and no errors have
#         occurred yet
#     Exception - an error has occurred in the cursor and should be raised
#         to the user once all results in `items` have been returned.  This
#         will be a StopIteration exception if the cursor completed successfully.
#
# A class that derives from this should implement the following function:
#     def _get_next(self, timeout):
# where `timeout` is the maximum amount of time (in seconds) to wait for the
# next result in the cursor before raising a RqlTimeoutError.
class Cursor(object):
    def __init__(self, conn_instance, query):
        self.conn = conn_instance
        self.query = query
        self.items = list()
        self.outstanding_requests = 1
        self.threshold = 0
        self.error = None
        self.it = iter(self._it())

        self.conn._cursor_cache[self.query.token] = self

    def close(self):
        if self.error is None:
            self.error = StopIteration()
            if self.conn.is_open():
                self.outstanding_requests += 1
                self.conn._parent._stop(self)

    def next(self, wait=True):
        if isinstance(wait, bool):
            timeout = None if wait else 0
        elif isinstance(wait, numbers.Real) and wait >= 0:
            timeout = wait
        else:
            raise RqlDriverError("Invalid wait timeout '%s'" % str(wait))
        return self._get_next(timeout)

    def _extend(self, res):
        self.outstanding_requests -= 1
        self.threshold = len(res.data)
        if self.error is None:
            if res.type == pResponse.SUCCESS_PARTIAL:
                self.items.extend(res.data)
            elif res.type == pResponse.SUCCESS_SEQUENCE:
                self.items.extend(res.data)
                self.error = StopIteration()
            else:
                self.error = res.make_error(self.query)
        self._maybe_fetch_batch()

        if self.outstanding_requests == 0 and self.error is not None:
            del self.conn._cursor_cache[res.token]

    def __iter__(self):
        return self

    def __next__(self):
        return next(self.it)<commit_msg>adding pretty-printed string conversion for python driver Cursors<commit_after># Copyright 2010-2014 RethinkDB, all rights reserved.

import errno
import json
import numbers
import socket
import struct
import time
try:
    from importlib import import_module
except ImportError:
    def import_module(name, package=None):
        # song & dance needed to do relative import in 2.6, which
        # doesn't have importlib
        return __import__(name[1:], globals(), locals(), [], 1)

from . import ql2_pb2 as p

__all__ = ['connect', 'set_loop_type', 'Connection', 'Cursor']

pResponse = p.Response.ResponseType
pQuery = p.Query.QueryType

from . import repl              # For the repl connection
from .errors import *
from .ast import RqlQuery, RqlTopLevelQuery, DB
from .ast import recursively_convert_pseudotypes

try:
    xrange
except NameError:
    xrange = range

try:
    {}.iteritems
    dict_items = lambda d: d.iteritems()
except AttributeError:
    dict_items = lambda d: d.items()

def decodeUTF(inputPipe):
    # attempt to decode input as utf-8 with fallbacks to get something useful
    try:
        return inputPipe.decode('utf-8', errors='ignore')
    except TypeError:
        try:
            return inputPipe.decode('utf-8')
        except UnicodeError:
            return repr(inputPipe)

def convert_pseudo(value, query):
    return recursively_convert_pseudotypes(value, query.global_optargs)

def maybe_profile(value, res):
    if res.profile is not None:
        return {'value': value, 'profile': res.profile}
    return value

class Query(object):
    def __init__(self, type, token, term, global_optargs):
        self.type = type
        self.token = token
        self.term = term
        self.global_optargs = global_optargs

    def serialize(self):
        message = [self.type]
        if self.term is not None:
            message.append(self.term.build())
        if self.global_optargs is not None:
            optargs = {}
            for k, v in dict_items(self.global_optargs):
                optargs[k] = v.build() if isinstance(v, RqlQuery) else v
            message.append(optargs)
        query_str = json.dumps(message, ensure_ascii=False, allow_nan=False).encode('utf-8')
        query_header = struct.pack('<QL', self.token, len(query_str))
        return query_header + query_str


class Response(object):
    def __init__(self, token, json_str):
        try:
            json_str = json_str.decode('utf-8')
        except AttributeError:
            pass               # Python3 str objects are already utf-8
        self.token = token
        full_response = json.loads(json_str)
        self.type = full_response["t"]
        self.data = full_response["r"]
        self.backtrace = full_response.get("b", None)
        self.profile = full_response.get("p", None)

    def make_error(self, query):
        if self.type == pResponse.CLIENT_ERROR:
            return RqlClientError(self.data[0], query.term, self.backtrace)
        elif self.type == pResponse.COMPILE_ERROR:
            return RqlCompileError(self.data[0], query.term, self.backtrace)
        elif self.type == pResponse.RUNTIME_ERROR:
            return RqlRuntimeError(self.data[0], query.term, self.backtrace)
        return RqlDriverError("Unknown Response type %d encountered" +
                              " in a response." % self.type)


# This class encapsulates all shared behavior between cursor implementations.
# It provides iteration over the cursor using `iter`, as well as incremental
# iteration using `next`.
#
# query - the original query that resulted in the cursor, used for:
#     query.term - the term to be used for pretty-printing backtraces
#     query.token - the token to use for subsequent CONTINUE and STOP requests
#     query.global_optargs - dictate how to format results
# items - The current list of items obtained from the server, this is
#     added to in `_extend`, which is called by the ConnectionInstance when a
#     new response arrives for this cursor.
# outstanding_requests - The number of requests that are currently awaiting
#     a response from the server.  This will typically be 0 or 1 unless the
#     cursor is exhausted, but this can be higher if `close` is called.
# threshold - a CONTINUE request will be sent when the length of `items` goes
#     below this number.
# error - indicates the current state of the cursor:
#     None - there is more data available from the server and no errors have
#         occurred yet
#     Exception - an error has occurred in the cursor and should be raised
#         to the user once all results in `items` have been returned.  This
#         will be a StopIteration exception if the cursor completed successfully.
#
# A class that derives from this should implement the following function:
#     def _get_next(self, timeout):
# where `timeout` is the maximum amount of time (in seconds) to wait for the
# next result in the cursor before raising a RqlTimeoutError.
class Cursor(object):
    def __init__(self, conn_instance, query):
        self.conn = conn_instance
        self.query = query
        self.items = list()
        self.outstanding_requests = 1
        self.threshold = 0
        self.error = None
        self.it = iter(self._it())

        self.conn._cursor_cache[self.query.token] = self

    def close(self):
        if self.error is None:
            self.error = StopIteration()
            if self.conn.is_open():
                self.outstanding_requests += 1
                self.conn._parent._stop(self)

    def next(self, wait=True):
        if isinstance(wait, bool):
            timeout = None if wait else 0
        elif isinstance(wait, numbers.Real) and wait >= 0:
            timeout = wait
        else:
            raise RqlDriverError("Invalid wait timeout '%s'" % str(wait))
        return self._get_next(timeout)

    def _extend(self, res):
        self.outstanding_requests -= 1
        self.threshold = len(res.data)
        if self.error is None:
            if res.type == pResponse.SUCCESS_PARTIAL:
                self.items.extend(res.data)
            elif res.type == pResponse.SUCCESS_SEQUENCE:
                self.items.extend(res.data)
                self.error = StopIteration()
            else:
                self.error = res.make_error(self.query)
        self._maybe_fetch_batch()

        if self.outstanding_requests == 0 and self.error is not None:
            del self.conn._cursor_cache[res.token]

    def __str__(self):
        val_str = ', '.join(map(repr, self.items[:10]))
        if len(self.items) > 10 or self.error is None:
            val_str += ', ...'

        if self.error is None:
            err_str = 'streaming'
        elif isinstance(self.error, StopIteration):
            err_str = 'done streaming'
        else:
            err_str = 'error: %s' % repr(self.error)

        return "%s (%s):\n[%s]" % (object.__str__(self), err_str, val_str)

    def __iter__(self):
        return self

    def __next__(self):
        return next(self.it)
{'fraction_non_alphanumeric': 0.08812260536398467, 'fraction_numerical': 0.005157677571470675, 'mean_word_length': 2.5318761384335153, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2466675', 'n_tokens_mistral': 3848, 'n_tokens_neox': 3735, 'n_words': 1266}