cursor.py 6.2 KB
Newer Older
1 2
from .cinterface import CTaosInterface
from .error import *
L
liuyq-617 已提交
3 4 5 6 7
from .constants import FieldType
import threading

# querySeqNum = 0

8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38

class TDengineCursor(object):
    """Database cursor which is used to manage the context of a fetch operation.

    Attributes:
        .description: Read-only attribute consists of 7-item sequences:

            > name (mondatory)
            > type_code (mondatory)
            > display_size
            > internal_size
            > precision
            > scale
            > null_ok

            This attribute will be None for operations that do not return rows or
            if the cursor has not had an operation invoked via the .execute*() method yet.

        .rowcount:This read-only attribute specifies the number of rows that the last
            .execute*() produced (for DQL statements like SELECT) or affected 
    """

    def __init__(self, connection=None):
        self._description = None
        self._rowcount = -1
        self._connection = None
        self._result = None
        self._fields = None
        self._block = None
        self._block_rows = -1
        self._block_iter = 0
39
        self._affected_rows = 0
L
liuyq-617 已提交
40 41
        self._logfile = ""
        self._threadId = threading.get_ident()
42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76

        if connection is not None:
            self._connection = connection

    def __iter__(self):
        return self

    def __next__(self):
        if self._result is None or self._fields is None:
            raise OperationalError("Invalid use of fetch iterator")

        if self._block_rows <= self._block_iter:
            block, self._block_rows = CTaosInterface.fetchBlock(self._result, self._fields)
            if self._block_rows == 0:
                raise StopIteration
            self._block = list(map(tuple, zip(*block)))
            self._block_iter = 0

        data =  self._block[self._block_iter]
        self._block_iter += 1

        return data

    @property
    def description(self):
        """Return the description of the object.
        """
        return self._description

    @property
    def rowcount(self):
        """Return the rowcount of the object
        """
        return self._rowcount

77 78 79 80 81 82
    @property
    def affected_rows(self):
        """Return the affected_rows of the object
        """
        return self._affected_rows

83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116
    def callproc(self, procname, *args):
        """Call a stored database procedure with the given name.

        Void functionality since no stored procedures.
        """
        pass

    def close(self):
        """Close the cursor.
        """
        if self._connection is None:
            return False
        
        self._reset_result()
        self._connection = None

        return True

    def execute(self, operation, params=None):
        """Prepare and execute a database operation (query or command).
        """
        if not operation:
            return None

        if not self._connection:
            # TODO : change the exception raised here
            raise ProgrammingError("Cursor is not connected")
        
        self._reset_result()

        stmt = operation
        if params is not None:
            pass
        
117
        self._result = CTaosInterface.query(self._connection._conn, stmt)
T
Tao Liu 已提交
118 119
        errno = CTaosInterface.libtaos.taos_errno(self._result)
        if errno == 0:
120 121 122
            if CTaosInterface.fieldsCount(self._result) == 0:
                self._affected_rows += CTaosInterface.affectedRows(self._result )
                return CTaosInterface.affectedRows(self._result )
123
            else:
124
                self._fields = CTaosInterface.useResult(self._result )
125 126
                return self._handle_result()
        else:
127
            raise ProgrammingError(CTaosInterface.errStr(self._result ), errno)
128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151

    def executemany(self, operation, seq_of_parameters):
        """Prepare a database operation (query or command) and then execute it against all parameter sequences or mappings found in the sequence seq_of_parameters.
        """
        pass

    def fetchone(self):
        """Fetch the next row of a query result set, returning a single sequence, or None when no more data is available.
        """
        pass

    def fetchmany(self):
        pass

    def fetchall(self):
        """Fetch all (remaining) rows of a query result, returning them as a sequence of sequences (e.g. a list of tuples). Note that the cursor's arraysize attribute can affect the performance of this operation.
        """
        if self._result is None or self._fields is None:
            raise OperationalError("Invalid use of fetchall")
        
        buffer = [[] for i in range(len(self._fields))]
        self._rowcount = 0
        while True:
            block, num_of_fields = CTaosInterface.fetchBlock(self._result, self._fields)
B
Bomin Zhang 已提交
152 153 154
            errno = CTaosInterface.libtaos.taos_errno(self._result)
            if errno != 0:
                raise ProgrammingError(CTaosInterface.errStr(self._result), errno)
155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180
            if num_of_fields == 0: break
            self._rowcount += num_of_fields
            for i in range(len(self._fields)):
                buffer[i].extend(block[i])

        
        return list(map(tuple, zip(*buffer)))



    def nextset(self):
        """
        """
        pass

    def setinputsize(self, sizes):
        pass

    def setutputsize(self, size, column=None):
        pass

    def _reset_result(self):
        """Reset the result to unused version.
        """
        self._description = None
        self._rowcount = -1
T
Tao Liu 已提交
181 182
        if self._result is not None:
            CTaosInterface.freeResult(self._result)
183 184 185 186 187
        self._result = None
        self._fields = None
        self._block = None
        self._block_rows = -1
        self._block_iter = 0
188
        self._affected_rows = 0
189 190 191 192 193 194 195 196
    
    def _handle_result(self):
        """Handle the return result from query.
        """
        self._description = []
        for ele in self._fields:
            self._description.append((ele['name'], ele['type'], None, None, None, None, False))
        
197
        return self._result