You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jb...@apache.org on 2011/04/13 16:03:43 UTC
svn commit: r1091789 - in /cassandra/branches/cassandra-0.8: drivers/py/
drivers/py/cql/ drivers/py/test/ test/system/
Author: jbellis
Date: Wed Apr 13 14:03:43 2011
New Revision: 1091789
URL: http://svn.apache.org/viewvc?rev=1091789&view=rev
Log:
update python driver to conform to PEP-249
patch by jhermes and thobbs; reviewed by jbellis for CASSANDRA-2402
Added:
cassandra/branches/cassandra-0.8/drivers/py/cql/cursor.py
Modified:
cassandra/branches/cassandra-0.8/drivers/py/README
cassandra/branches/cassandra-0.8/drivers/py/cql/__init__.py
cassandra/branches/cassandra-0.8/drivers/py/cql/connection.py
cassandra/branches/cassandra-0.8/drivers/py/cql/decoders.py
cassandra/branches/cassandra-0.8/drivers/py/cql/marshal.py
cassandra/branches/cassandra-0.8/drivers/py/cql/results.py
cassandra/branches/cassandra-0.8/drivers/py/test/test_query_compression.py
cassandra/branches/cassandra-0.8/drivers/py/test/test_query_preparation.py
cassandra/branches/cassandra-0.8/test/system/test_cql.py
Modified: cassandra/branches/cassandra-0.8/drivers/py/README
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/drivers/py/README?rev=1091789&r1=1091788&r2=1091789&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/drivers/py/README (original)
+++ cassandra/branches/cassandra-0.8/drivers/py/README Wed Apr 13 14:03:43 2011
@@ -1,2 +1,28 @@
-Python language driver for CQL, a SQL-alike query language for Apache
-Cassandra.
+A Python driver for CQL that adheres to py-dbapi v2
+ (PEP249, Python Database API Specification v2.0: http://www.python.org/dev/peps/pep-0249/).
+
+Standard use:
+ >> import cql
+ >> con = cql.connect(host, port, keyspace)
+ >> cursor = con.cursor()
+ >> cursor.execute("CQL QUERY", {kw=Foo, kw2=Bar, etc...})
+
+ - cursor.description # None initially, list of N tuples that represent
+ the N columns in a row after an execute. Only
+ contains type and name info, not values.
+ - cursor.rowcount # -1 initially, N after an execute
+ - cursor.arraysize # variable size of a fetchmany call
+ - cursor.fetchone() # returns a single row
+ - cursor.fetchmany() # returns self.arraysize # of rows
+ - cursor.fetchall() # returns all rows, don't do this.
+
+ >> cursor.execute("ANOTHER QUERY", **more_kwargs)
+ >> for row in cursor: # Iteration is equivalent to lots of fetchone() calls
+ >> doRowMagic(row)
+
+ >> cursor.close()
+ >> con.close()
+
+Query substitution:
+ - Use named parameters and a dictionary of names and values.
+ e.g. execute("SELECT * FROM CF WHERE name=:name", name="Foo")
Modified: cassandra/branches/cassandra-0.8/drivers/py/cql/__init__.py
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/drivers/py/cql/__init__.py?rev=1091789&r1=1091788&r2=1091789&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/drivers/py/cql/__init__.py (original)
+++ cassandra/branches/cassandra-0.8/drivers/py/cql/__init__.py Wed Apr 13 14:03:43 2011
@@ -15,9 +15,82 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-"""
-Cassandra Query Language driver
-"""
+import exceptions
+import datetime
+
+import connection
+import marshal
+
+
+# dbapi Error hierarchy
+
+class Warning(exceptions.StandardError): pass
+class Error (exceptions.StandardError): pass
+
+class InterfaceError(Error): pass
+class DatabaseError (Error): pass
+
+class DataError (DatabaseError): pass
+class OperationalError (DatabaseError): pass
+class IntegrityError (DatabaseError): pass
+class InternalError (DatabaseError): pass
+class ProgrammingError (DatabaseError): pass
+class NotSupportedError(DatabaseError): pass
+
+
+# Module constants
+
+apilevel = 1.0
+threadsafety = 1 # Threads may share the module, but not connections/cursors.
+paramstyle = 'named'
+
+ROW_KEY = "Row Key"
+
+# TODO: Pull connections out of a pool instead.
+def connect(host, port=9160, keyspace='system', user=None, password=None):
+ return connection.Connection(host, port, keyspace, user, password)
+
+# Module Type Objects and Constructors
+
+Date = datetime.date
+
+Time = datetime.time
+
+Timestamp = datetime.datetime
+
+Binary = buffer
+
+def DateFromTicks(ticks):
+ return Date(*time.localtime(ticks)[:3])
+
+def TimeFromTicks(ticks):
+ return Time(*time.localtime(ticks)[3:6])
+
+def TimestampFromTicks(ticks):
+ return Timestamp(*time.localtime(ticks)[:6])
+
+class DBAPITypeObject:
+
+ def __init__(self, *values):
+ self.values = values
+
+ def __cmp__(self,other):
+ if other in self.values:
+ return 0
+ if other < self.values:
+ return 1
+ else:
+ return -1
+
+STRING = DBAPITypeObject(marshal.BYTES_TYPE, marshal.ASCII_TYPE, marshal.UTF8_TYPE)
+
+BINARY = DBAPITypeObject(marshal.BYTES_TYPE, marshal.UUID_TYPE, marshal.LEXICAL_UUID_TYPE)
+
+NUMBER = DBAPITypeObject(marshal.LONG_TYPE, marshal.INTEGER_TYPE)
+
+DATETIME = DBAPITypeObject(marshal.TIME_UUID_TYPE)
+
+ROWID = DBAPITypeObject(marshal.BYTES_TYPE, marshal.ASCII_TYPE, marshal.UTF8_TYPE,
+ marshal.INTEGER_TYPE, marshal.LONG_TYPE, marshal.UUID_TYPE,
+ marshal.LEXICAL_UUID_TYPE, marshal.TIME_UUID_TYPE)
-from connection import Connection
-from connection_pool import ConnectionPool
Modified: cassandra/branches/cassandra-0.8/drivers/py/cql/connection.py
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/drivers/py/cql/connection.py?rev=1091789&r1=1091788&r2=1091789&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/drivers/py/cql/connection.py (original)
+++ cassandra/branches/cassandra-0.8/drivers/py/cql/connection.py Wed Apr 13 14:03:43 2011
@@ -15,183 +15,69 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from os.path import exists, abspath, dirname, join
+from cursor import Cursor
+from cassandra import Cassandra
from thrift.transport import TTransport, TSocket
from thrift.protocol import TBinaryProtocol
-from thrift.Thrift import TApplicationException
-from errors import CQLException, InvalidCompressionScheme
-from marshal import prepare
-from decoders import SchemaDecoder
-from results import RowsProxy
-import zlib, re
-
-from cassandra import Cassandra
-from cassandra.ttypes import Compression, InvalidRequestException, \
- CqlResultType, AuthenticationRequest, \
- SchemaDisagreementException
-
-COMPRESSION_SCHEMES = ['GZIP', 'NONE']
-DEFAULT_COMPRESSION = 'GZIP'
-__all__ = ['COMPRESSION_SCHEMES', 'DEFAULT_COMPRESSION', 'Connection']
class Connection(object):
- """
- CQL connection object.
-
- Example usage:
- >>> conn = Connection("localhost", keyspace="Keyspace1")
- >>> r = conn.execute('SELECT "age" FROM Users')
- >>> for row in r.rows:
- ... for column in row.columns:
- ... print "%s is %s years of age" % (r.key, column.age)
- """
- _keyspace_re = re.compile("USE (\w+);?", re.I | re.M)
- _cfamily_re = re.compile("\s*SELECT\s+.+\s+FROM\s+[\']?(\w+)", re.I | re.M)
- _ddl_re = re.compile("\s*(CREATE|ALTER|DROP)\s+", re.I | re.M)
-
- def __init__(self, host, port=9160, keyspace=None, username=None,
- password=None, decoder=None):
+
+ def __init__(self, host, port, keyspace, user=None, password=None):
"""
Params:
* host .........: hostname of Cassandra node.
- * port .........: port number to connect to (optional).
- * keyspace .....: keyspace name (optional).
- * username .....: username used in authentication (optional).
+ * port .........: port number to connect to.
+ * keyspace .....: keyspace to connect to.
+ * user .........: username used in authentication (optional).
* password .....: password used in authentication (optional).
- * decoder ......: result decoder instance (optional, defaults to none).
"""
+ self.host = host
+ self.port = port
+ self.keyspace = keyspace
+
socket = TSocket.TSocket(host, port)
self.transport = TTransport.TFramedTransport(socket)
protocol = TBinaryProtocol.TBinaryProtocolAccelerated(self.transport)
self.client = Cassandra.Client(protocol)
+
socket.open()
-
- # XXX: "current" is probably a misnomer.
- self._cur_keyspace = None
- self._cur_column_family = None
-
- if username and password:
- credentials = {"username": username, "password": password}
- self.client.login(AuthenticationRequest(credentials=credentials))
-
- if keyspace:
- self.execute('USE %s;' % keyspace)
- self._cur_keyspace = keyspace
-
- if not decoder:
- self.decoder = SchemaDecoder(self.__get_schema())
- else:
- self.decoder = decoder
-
- def __get_schema(self):
- def columns(metadata):
- results = {}
- for col in metadata:
- results[col.name] = col.validation_class
- return results
-
- def column_families(cf_defs):
- cfresults = {}
- for cf in cf_defs:
- cfresults[cf.name] = {"comparator": cf.comparator_type}
- cfresults[cf.name]["default_validation_class"] = \
- cf.default_validation_class
- cfresults[cf.name]["key_validation_class"] = \
- cf.key_validation_class
- cfresults[cf.name]["columns"] = columns(cf.column_metadata)
- return cfresults
-
- schema = {}
- for ksdef in self.client.describe_keyspaces():
- schema[ksdef.name] = column_families(ksdef.cf_defs)
- return schema
-
- def prepare(self, query, *args):
- prepared_query = prepare(query, *args)
-
- # Snag the keyspace or column family and stash it for later use in
- # decoding columns. These regexes don't match every query, but the
- # current column family only needs to be current for SELECTs.
- match = Connection._cfamily_re.match(prepared_query)
- if match:
- self._cur_column_family = match.group(1)
- return prepared_query
- match = Connection._keyspace_re.match(prepared_query)
- if match:
- self._cur_keyspace = match.group(1)
- return prepared_query
-
- # If this is a CREATE, then refresh the schema for decoding purposes.
- match = Connection._ddl_re.match(prepared_query)
- if match:
- if isinstance(self.decoder, SchemaDecoder):
- self.decoder.schema = self.__get_schema()
-
- return prepared_query
+ self.open_socket = True
- def execute(self, query, *args, **kwargs):
- """
- Execute a CQL query on a remote node.
-
- Params:
- * query .........: CQL query string.
- * args ..........: Query parameters.
- * compression ...: Query compression type (optional).
- """
- if kwargs.has_key("compression"):
- compress = kwargs.get("compression").upper()
- else:
- compress = DEFAULT_COMPRESSION
-
- compressed_query = Connection.compress_query(self.prepare(query, *args),
- compress)
- request_compression = getattr(Compression, compress)
-
- try:
- response = self.client.execute_cql_query(compressed_query,
- request_compression)
- except InvalidRequestException, ire:
- raise CQLException("Bad Request: %s" % ire.why)
- except SchemaDisagreementException, sde:
- raise CQLException("schema versions disagree, (try again later).")
- except TApplicationException, tapp:
- raise CQLException("Internal application error")
- except Exception, exc:
- raise CQLException(exc)
-
- if response.type == CqlResultType.ROWS:
- return RowsProxy(response.rows,
- self._cur_keyspace,
- self._cur_column_family,
- self.decoder)
-
- if response.type == CqlResultType.INT:
- return response.num
+ if user and password:
+ credentials = {"username": user, "password": password}
+ self.client.login(AuthenticationRequest(credentials=credentials))
- return None
+ if keyspace:
+ c = self.cursor()
+ c.execute('USE %s;' % keyspace)
+ c.close()
+
+ def __str__(self):
+ return "{host: '%s:%s', keyspace: '%s'}"%(self.host,self.port,self.keyspace)
+
+ ###
+ # Connection API
+ ###
def close(self):
+ if not self.open_socket:
+ raise InternalError("Connection has been closed.")
+
self.transport.close()
-
- def is_open(self):
- return self.transport.isOpen()
+ self.open_socket = False
- @classmethod
- def compress_query(cls, query, compression):
+ def commit(self):
"""
- Returns a query string compressed with the specified compression type.
-
- Params:
- * query .........: The query string to compress.
- * compression ...: Type of compression to use.
+ 'Database modules that do not support transactions should
+ implement this method with void functionality.'
"""
- if not compression in COMPRESSION_SCHEMES:
- raise InvalidCompressionScheme(compression)
+ return
+
+ def rollback(self):
+ raise NotSupportedError("Rollback functionality not present in Cassandra.")
- if compression == 'GZIP':
- return zlib.compress(query)
-
- return query
-
-# vi: ai ts=4 tw=0 sw=4 et
+ def cursor(self):
+ if not self.open_socket:
+ raise InternalError("Connection has been closed.")
+ return Cursor(self)
Added: cassandra/branches/cassandra-0.8/drivers/py/cql/cursor.py
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/drivers/py/cql/cursor.py?rev=1091789&view=auto
==============================================================================
--- cassandra/branches/cassandra-0.8/drivers/py/cql/cursor.py (added)
+++ cassandra/branches/cassandra-0.8/drivers/py/cql/cursor.py Wed Apr 13 14:03:43 2011
@@ -0,0 +1,225 @@
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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 re
+import zlib
+
+import cql
+from cql.marshal import prepare
+from cql.decoders import SchemaDecoder
+from cql.results import ResultSet
+from cql.cassandra.ttypes import (Compression, CqlResultType, InvalidRequestException,
+ TApplicationException)
+
+class Cursor:
+
+ _keyspace_re = re.compile("USE (\w+);?", re.I | re.M)
+ _cfamily_re = re.compile("\s*SELECT\s+.+\s+FROM\s+[\']?(\w+)", re.I | re.M)
+ _ddl_re = re.compile("\s*(CREATE|ALTER|DROP)\s+", re.I | re.M)
+
+ def __init__(self, parent_connection):
+ self.open_socket = True
+ self.parent_connection = parent_connection
+
+ self.result = None # Populate on execute()
+ self.description = None # A list of 7-tuples:
+ # (column_name, type_code, none, none,
+ # none, none, nulls_ok=True)
+ # Populate on execute()
+
+ self.arraysize = 1
+ self.rowcount = -1 # Populate on execute()
+ self.compression = 'GZIP'
+
+ self._query_ks = self.parent_connection.keyspace
+ self._query_cf = None
+ self.decoder = SchemaDecoder(self.__get_schema())
+
+ ###
+ # Cursor API
+ ###
+
+ def close(self):
+ self.open_socket = False
+
+ def prepare(self, query, params):
+ prepared_query = prepare(query, params)
+
+ # Snag the keyspace or column family and stash it for later use in
+ # decoding columns. These regexes don't match every query, but the
+ # current column family only needs to be current for SELECTs.
+ match = Cursor._cfamily_re.match(prepared_query)
+ if match:
+ self._query_cf = match.group(1)
+ return prepared_query
+ match = Cursor._keyspace_re.match(prepared_query)
+ if match:
+ self._query_ks = match.group(1)
+ return prepared_query
+
+ # If this is a CREATE, then refresh the schema for decoding purposes.
+ match = Cursor._ddl_re.match(prepared_query)
+ if match:
+ if isinstance(self.decoder, SchemaDecoder):
+ self.decoder.schema = self.__get_schema()
+
+ return prepared_query
+
+ def __get_schema(self):
+ def columns(metadata):
+ results = {}
+ for col in metadata:
+ results[col.name] = col.validation_class
+ return results
+
+ def column_families(cf_defs):
+ cfresults = {}
+ if cf_defs:
+ for cf in cf_defs:
+ cfresults[cf.name] = {"comparator": cf.comparator_type}
+ cfresults[cf.name]["default_validation_class"] = \
+ cf.default_validation_class
+ cfresults[cf.name]["key_validation_class"] = \
+ cf.key_validation_class
+ cfresults[cf.name]["columns"] = columns(cf.column_metadata)
+ return cfresults
+
+ schema = {}
+ client = self.parent_connection.client
+ for ksdef in client.describe_keyspaces():
+ schema[ksdef.name] = column_families(ksdef.cf_defs)
+ return schema
+
+ def execute(self, cql_query, params={}):
+ self.__checksock()
+ try:
+ prepared_q = self.prepare(cql_query, params)
+ except KeyError, e:
+ raise cql.ProgrammingError("Unmatched named substitution: " +
+ "%s not given for %s" % (e, cql_query))
+
+ if self.compression == 'GZIP':
+ compressed_q = zlib.compress(prepared_q)
+ else:
+ compressed_q = prepared_q
+ request_compression = getattr(Compression, self.compression)
+
+ try:
+ client = self.parent_connection.client
+ response = client.execute_cql_query(compressed_q, request_compression)
+ except InvalidRequestException, ire:
+ raise cql.ProgrammingError("Bad Request: %s" % ire.why)
+ except SchemaDisagreementException, sde:
+ raise cql.IntegrityError("Schema versions disagree, (try again later).")
+ except TApplicationException, tapp:
+ raise cql.InternalError("Internal application error")
+
+ if response.type == CqlResultType.ROWS:
+ self.result = ResultSet(response.rows,
+ self._query_ks,
+ self._query_cf,
+ self.decoder)
+ self.rs_idx = 0
+ self.rowcount = len(self.result)
+ self.description = self.result.description
+
+ if response.type == CqlResultType.INT:
+ self.result = [(response.num,)]
+ self.rs_idx = 0
+ self.rowcount = 1
+ # TODO: name could be the COUNT expression
+ self.description = (None, None, None, None, None, None, None)
+
+ # 'Return values are not defined.'
+ return True
+
+ def executemany(self, operation_list, argslist):
+ self.__checksock()
+ opssize = len(operation_list)
+ argsize = len(argslist)
+
+ if opssize > argsize:
+ raise cql.InterfaceError("Operations outnumber args for executemany().")
+ elif opssize < argsize:
+ raise cql.InterfaceError("Args outnumber operations for executemany().")
+
+ for idx in xrange(opssize):
+ self.execute(operation_list[idx], *argslist[idx])
+
+ def fetchone(self):
+ self.__checksock()
+ ret = self.result[self.rs_idx]
+ self.rs_idx += 1
+ self.description = getattr(self.result, 'description', self.description)
+ return ret
+
+ def fetchmany(self, size=None):
+ self.__checksock()
+ if size is None:
+ size = self.arraysize
+ end = self.rs_idx + size
+ ret = self.result[self.rs_idx:end]
+ self.rs_idx = end
+ self.description = getattr(self.result, 'description', self.description)
+ return ret
+
+ def fetchall(self):
+ self.__checksock()
+ ret = self.result[self.rs_idx:]
+ self.rs_idx = len(self.result)
+ self.description = self.result.description
+ return ret
+
+ ###
+ # Iterator extension
+ ###
+
+ def next(self):
+ raise Warning("DB-API extension cursor.next() used")
+
+ if self.rs_idx >= len(self.result):
+ raise StopIteration
+ return self.fetchone()
+
+ def __iter__(self):
+ raise Warning("DB-API extension cursor.__iter__() used")
+ return self
+
+ ###
+ # Unsupported, unimplemented optionally
+ ###
+
+ def setinputsizes(self, sizes):
+ pass # DO NOTHING
+
+ def setoutputsize(self, size, *columns):
+ pass # DO NOTHING
+
+ def callproc(self, procname, *args):
+ raise cql.NotSupportedError()
+
+ def nextset(self):
+ raise cql.NotSupportedError()
+
+ ###
+ # Helpers
+ ###
+
+ def __checksock(self):
+ if not self.open_socket:
+ raise cql.InternalError("Cursor belonging to %s has been closed." %
+ (self.parent_connection, ))
Modified: cassandra/branches/cassandra-0.8/drivers/py/cql/decoders.py
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/drivers/py/cql/decoders.py?rev=1091789&r1=1091788&r2=1091789&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/drivers/py/cql/decoders.py (original)
+++ cassandra/branches/cassandra-0.8/drivers/py/cql/decoders.py Wed Apr 13 14:03:43 2011
@@ -15,62 +15,53 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from os.path import abspath, dirname, join
-from marshal import unmarshal
+import cql
+from marshal import (unmarshallers, unmarshal_noop)
-class BaseDecoder(object):
- def decode_column(self, keyspace, column_family, name, value):
- raise NotImplementedError()
-
- def decode_key(self, keyspace, column_family, key):
- raise NotImplementedError()
-
-class NoopDecoder(BaseDecoder):
- def decode_column(self, keyspace, column_family, name, value):
- return (name, value)
-
- def decode_key(self, keyspace, column_family, key):
- return key
-
-class SchemaDecoder(BaseDecoder):
+class SchemaDecoder(object):
"""
Decode binary column names/values according to schema.
"""
def __init__(self, schema={}):
self.schema = schema
-
+
def __get_column_family_def(self, keyspace, column_family):
if self.schema.has_key(keyspace):
if self.schema[keyspace].has_key(column_family):
return self.schema[keyspace][column_family]
return None
-
+
def __comparator_for(self, keyspace, column_family):
cfam = self.__get_column_family_def(keyspace, column_family)
- if cfam and cfam.has_key("comparator"):
+ if cfam and "comparator" in cfam:
return cfam["comparator"]
return None
-
+
def __validator_for(self, keyspace, column_family, name):
cfam = self.__get_column_family_def(keyspace, column_family)
if cfam:
- if cfam["columns"].has_key(name):
+ if name in cfam["columns"]:
return cfam["columns"][name]
- else:
- return cfam["default_validation_class"]
+ return cfam["default_validation_class"]
return None
-
- def __keytype_for(self, keyspace, column_family, key):
+
+ def __keytype_for(self, keyspace, column_family):
cfam = self.__get_column_family_def(keyspace, column_family)
- if cfam and cfam.has_key("key_validation_class"):
+ if cfam and "key_validation_class" in cfam:
return cfam["key_validation_class"]
return None
- def decode_column(self, keyspace, column_family, name, value):
+ def decode_row(self, keyspace, column_family, row):
+ key_type = self.__keytype_for(keyspace, column_family)
+ key = unmarshallers.get(key_type, unmarshal_noop)(row.key)
+ description = [(cql.ROW_KEY, key_type, None, None, None, None, None, False)]
+
comparator = self.__comparator_for(keyspace, column_family)
- validator = self.__validator_for(keyspace, column_family, name)
- return (unmarshal(name, comparator), unmarshal(value, validator))
-
- def decode_key(self, keyspace, column_family, key):
- return unmarshal(key, self.__keytype_for(keyspace, column_family, key))
-
+ unmarshal = unmarshallers.get(comparator, unmarshal_noop)
+ values = [key]
+ for column in row.columns:
+ description.append((unmarshal(column.name), comparator, None, None, None, None, True))
+ validator = self.__validator_for(keyspace, column_family, column.name)
+ values.append(unmarshallers.get(validator, unmarshal_noop)(column.value))
+
+ return description, values
Modified: cassandra/branches/cassandra-0.8/drivers/py/cql/marshal.py
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/drivers/py/cql/marshal.py?rev=1091789&r1=1091788&r2=1091789&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/drivers/py/cql/marshal.py (original)
+++ cassandra/branches/cassandra-0.8/drivers/py/cql/marshal.py Wed Apr 13 14:03:43 2011
@@ -15,68 +15,75 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+import re
+import struct
from uuid import UUID
-from StringIO import StringIO
-from errors import InvalidQueryFormat
-from struct import unpack
-
-__all__ = ['prepare', 'marshal', 'unmarshal']
-
-def prepare(query, *args):
- result = StringIO()
- index = query.find('?')
- oldindex = 0
- count = 0
-
- while (index >= 0):
- result.write(query[oldindex:index])
- try:
- result.write(marshal(args[count]))
- except IndexError:
- raise InvalidQueryFormat("not enough arguments in substitution")
-
- oldindex = index + 1
- index = query.find('?', index + 1)
- count += 1
- result.write(query[oldindex:])
-
- if count < len(args):
- raise InvalidQueryFormat("too many arguments in substitution")
-
- return result.getvalue()
+
+import cql
+
+__all__ = ['prepare', 'marshal', 'unmarshal_noop', 'unmarshallers']
+
+if hasattr(struct, 'Struct'): # new in Python 2.5
+ _have_struct = True
+ _long_packer = struct.Struct('>q')
+else:
+ _have_struct = False
+
+_param_re = re.compile(r"(?<!strategy_options)(:[a-zA-Z_][a-zA-Z0-9_]*)", re.M)
+
+BYTES_TYPE = "org.apache.cassandra.db.marshal.BytesType"
+ASCII_TYPE = "org.apache.cassandra.db.marshal.AsciiType"
+UTF8_TYPE = "org.apache.cassandra.db.marshal.UTF8Type"
+INTEGER_TYPE = "org.apache.cassandra.db.marshal.IntegerType"
+LONG_TYPE = "org.apache.cassandra.db.marshal.LongType"
+UUID_TYPE = "org.apache.cassandra.db.marshal.UUIDType"
+LEXICAL_UUID_TYPE = "org.apache.cassandra.db.marshal.LexicalType"
+TIME_UUID_TYPE = "org.apache.cassandra.db.marshal.TimeUUIDType"
+
+def prepare(query, params):
+ # For every match of the form ":param_name", call marshal
+ # on kwargs['param_name'] and replace that section of the query
+ # with the result
+ new, count = re.subn(_param_re, lambda m: marshal(params[m.group(1)[1:]]), query)
+ if len(params) > count:
+ raise cql.ProgrammingError("More keywords were provided than parameters")
+ return new
def marshal(term):
- if isinstance(term, (long,int)):
- return "%d" % term
- elif isinstance(term, unicode):
+ if isinstance(term, unicode):
return "'%s'" % __escape_quotes(term.encode('utf8'))
elif isinstance(term, str):
return "'%s'" % __escape_quotes(term)
- elif isinstance(term, UUID):
+ else:
return str(term)
+
+def unmarshal_noop(bytestr):
+ return bytestr
+
+def unmarshal_utf8(bytestr):
+ return bytestr.decode("utf8")
+
+def unmarshal_int(bytestr):
+ return decode_bigint(bytestr)
+
+def unmarshal_long(bytestr):
+ if _have_struct:
+ return _long_packer.unpack(bytestr)[0]
else:
- return str(term)
-
-def unmarshal(bytestr, typestr):
- if typestr == "org.apache.cassandra.db.marshal.BytesType":
- return bytestr
- elif typestr == "org.apache.cassandra.db.marshal.AsciiType":
- return bytestr
- elif typestr == "org.apache.cassandra.db.marshal.UTF8Type":
- return bytestr.decode("utf8")
- elif typestr == "org.apache.cassandra.db.marshal.IntegerType":
- return decode_bigint(bytestr)
- elif typestr == "org.apache.cassandra.db.marshal.LongType":
return unpack(">q", bytestr)[0]
- elif typestr == "org.apache.cassandra.db.marshal.UUIDType":
- return UUID(bytes=bytestr)
- elif typestr == "org.apache.cassandra.db.marshal.LexicalUUIDType":
- return UUID(bytes=bytestr)
- elif typestr == "org.apache.cassandra.db.marshal.TimeUUIDType":
- return UUID(bytes=bytestr)
- else:
- return bytestr
-
+
+def unmarshal_uuid(bytestr):
+ return UUID(bytes=bytestr)
+
+unmarshallers = {BYTES_TYPE: unmarshal_noop,
+ ASCII_TYPE: unmarshal_noop,
+ UTF8_TYPE: unmarshal_utf8,
+ INTEGER_TYPE: unmarshal_int,
+ LONG_TYPE: unmarshal_long,
+ UUID_TYPE: unmarshal_uuid,
+ LEXICAL_UUID_TYPE: unmarshal_uuid,
+ TIME_UUID_TYPE: unmarshal_uuid}
+
def decode_bigint(term):
val = int(term.encode('hex'), 16)
if (ord(term[0]) & 128) != 0:
Modified: cassandra/branches/cassandra-0.8/drivers/py/cql/results.py
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/drivers/py/cql/results.py?rev=1091789&r1=1091788&r2=1091789&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/drivers/py/cql/results.py (original)
+++ cassandra/branches/cassandra-0.8/drivers/py/cql/results.py Wed Apr 13 14:03:43 2011
@@ -15,84 +15,48 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-class RowsProxy(object):
- def __init__(self, rows, keyspace, cfam, decoder):
+class ResultSet(object):
+
+ def __init__(self, rows, keyspace, column_family, decoder):
self.rows = rows
- self.keyspace = keyspace
- self.cfam = cfam
+ self.ks = keyspace
+ self.cf = column_family
self.decoder = decoder
- def __len__(self):
- return len(self.rows)
-
- def __getitem__(self, idx):
- return Row(self.rows[idx].key,
- self.rows[idx].columns,
- self.keyspace,
- self.cfam,
- self.decoder)
-
- def __iter__(self):
- for r in self.rows:
- yield Row(r.key, r.columns, self.keyspace, self.cfam, self.decoder)
+ # We need to try to parse the first row to set the description
+ if len(self.rows) > 0:
+ self.description, self._first_vals = self.decoder.decode_row(self.ks, self.cf, self.rows[0])
+ else:
+ self.description, self._first_vals = (None, None)
-class Row(object):
- def __init__(self, key, columns, keyspace, cfam, decoder):
- self._key = key
- self.keyspace = keyspace
- self.cfam = cfam
- self.decoder = decoder
- self.columns = ColumnsProxy(columns, keyspace, cfam, decoder)
-
- def __get_key(self):
- return self.decoder.decode_key(self.keyspace, self.cfam, self._key)
- key = property(__get_key)
-
- def __iter__(self):
- return iter(self.columns)
-
- def __getitem__(self, idx):
- return self.columns[idx]
-
def __len__(self):
- return len(self.columns)
+ return len(self.rows)
-class ColumnsProxy(object):
- def __init__(self, columns, keyspace, cfam, decoder):
- self.columns = columns
- self.keyspace = keyspace
- self.cfam = cfam
- self.decoder = decoder
-
- def __len__(self):
- return len(self.columns)
-
def __getitem__(self, idx):
- return Column(self.decoder.decode_column(self.keyspace,
- self.cfam,
- self.columns[idx].name,
- self.columns[idx].value))
+ if isinstance(idx, int):
+ if idx == 0 and self._first_vals:
+ return self._first_vals
+ self.description, vals = self.decoder.decode_row(self.ks, self.cf, self.rows[idx])
+ return vals
+ elif isinstance(idx, slice):
+ num_rows = len(self.rows)
+ results = []
+ for i in xrange(idx.start, min(len(self.rows), idx.stop)):
+ if i == 0 and self._first_vals:
+ vals = self._first_vals
+ else:
+ self.description, vals = self.decoder.decode_row(self.ks, self.cf, self.rows[i])
+ results.append(vals)
+ return results
+ else:
+ raise TypeError
def __iter__(self):
- for c in self.columns:
- yield Column(self.decoder.decode_column(self.keyspace,
- self.cfam,
- c.name,
- c.value))
-
- def __str__(self):
- return "ColumnsProxy(columns=%s)" % self.columns
-
- def __repr__(self):
- return str(self)
-
-class Column(object):
- def __init__(self, (name, value)):
- self.name = name
- self.value = value
-
- def __str__(self):
- return "Column(%s, %s)" % (self.name, self.value)
-
- def __repr__(self):
- return str(self)
+ idx = 0
+ for r in self.rows:
+ if idx == 0 and self._first_vals:
+ yield self._first_vals
+ else:
+ self.description, vals = self.decoder.decode_row(self.ks, self.cf, r)
+ yield vals
+ idx += 1
Modified: cassandra/branches/cassandra-0.8/drivers/py/test/test_query_compression.py
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/drivers/py/test/test_query_compression.py?rev=1091789&r1=1091788&r2=1091789&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/drivers/py/test/test_query_compression.py (original)
+++ cassandra/branches/cassandra-0.8/drivers/py/test/test_query_compression.py Wed Apr 13 14:03:43 2011
@@ -15,18 +15,12 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from os.path import abspath, exists, join, dirname
-
-if exists(join(abspath(dirname(__file__)), '..', 'cql')):
- import sys; sys.path.append(join(abspath(dirname(__file__)), '..'))
-
import unittest, zlib
-from cql import Connection
class TestCompression(unittest.TestCase):
def test_gzip(self):
"compressing a string w/ gzip"
query = "SELECT \"foo\" FROM Standard1 WHERE KEY = \"bar\";"
- compressed = Connection.compress_query(query, 'GZIP')
+ compressed = zlib.compress(query)
decompressed = zlib.decompress(compressed)
assert query == decompressed, "Decompressed query did not match"
Modified: cassandra/branches/cassandra-0.8/drivers/py/test/test_query_preparation.py
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/drivers/py/test/test_query_preparation.py?rev=1091789&r1=1091788&r2=1091789&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/drivers/py/test/test_query_preparation.py (original)
+++ cassandra/branches/cassandra-0.8/drivers/py/test/test_query_preparation.py Wed Apr 13 14:03:43 2011
@@ -16,26 +16,26 @@
# limitations under the License.
import unittest
+import cql
from cql.marshal import prepare
-from cql.errors import InvalidQueryFormat
# TESTS[i] ARGUMENTS[i] -> STANDARDS[i]
TESTS = (
"""
-SELECT ?,?,?,? FROM ColumnFamily WHERE KEY = ? AND 'col' = ?;
+SELECT :a,:b,:c,:d FROM ColumnFamily WHERE KEY = :e AND 'col' = :f;
""",
"""
USE Keyspace;
""",
"""
-SELECT ?..? FROM ColumnFamily;
+SELECT :a..:b FROM ColumnFamily;
""",
)
ARGUMENTS = (
- (1, 3, long(1000), long(3000), "key", unicode("val")),
- tuple(),
- ("a'b", "c'd'e"),
+ {'a': 1, 'b': 3, 'c': long(1000), 'd': long(3000), 'e': "key", 'f': unicode("val")},
+ {},
+ {'a': "a'b", 'b': "c'd'e"},
)
STANDARDS = (
@@ -54,13 +54,12 @@ class TestPrepare(unittest.TestCase):
def test_prepares(self):
"test prepared queries against known standards"
for (i, test) in enumerate(TESTS):
- a = prepare(test, *ARGUMENTS[i])
+ a = prepare(test, **ARGUMENTS[i])
b = STANDARDS[i]
assert a == b, "\n%s !=\n%s" % (a, b)
-
+
def test_bad(self):
"ensure bad calls raise exceptions"
- self.assertRaises(InvalidQueryFormat, prepare, "? ?", 1)
- self.assertRaises(InvalidQueryFormat, prepare, "? ?", 1, 2, 3)
- self.assertRaises(InvalidQueryFormat, prepare, "none", 1)
-
+ self.assertRaises(KeyError, prepare, ":a :b", a=1)
+ self.assertRaises(cql.ProgrammingError, prepare, ":a :b", a=1, b=2, c=3)
+ self.assertRaises(cql.ProgrammingError, prepare, "none", a=1)
Modified: cassandra/branches/cassandra-0.8/test/system/test_cql.py
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/test/system/test_cql.py?rev=1091789&r1=1091788&r2=1091789&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/test/system/test_cql.py (original)
+++ cassandra/branches/cassandra-0.8/test/system/test_cql.py Wed Apr 13 14:03:43 2011
@@ -23,18 +23,18 @@ import sys, uuid, time
sys.path.append(join(abspath(dirname(__file__)), '../../drivers/py'))
-from cql import Connection
-from cql.errors import CQLException
+import cql
+from cql.connection import Connection
from __init__ import ThriftTester
from __init__ import thrift_client # TODO: temporary
-def assert_raises(exception, method, *args):
+def assert_raises(exception, method, *args, **kwargs):
try:
- method(*args)
+ method(*args, **kwargs)
except exception:
return
raise AssertionError("failed to see expected exception")
-
+
def uuid1bytes_to_millis(uuidbytes):
return (uuid.UUID(bytes=uuidbytes).get_time() / 10000) - 12219292800000L
@@ -72,12 +72,12 @@ def load_sample(dbconn):
WITH comparator = ascii AND default_validation = ascii;
""")
dbconn.execute("CREATE INDEX ON IndexedA (birthdate)")
-
- query = "UPDATE StandardString1 SET ? = ?, ? = ? WHERE KEY = ?"
- dbconn.execute(query, "ca1", "va1", "col", "val", "ka")
- dbconn.execute(query, "cb1", "vb1", "col", "val", "kb")
- dbconn.execute(query, "cc1", "vc1", "col", "val", "kc")
- dbconn.execute(query, "cd1", "vd1", "col", "val", "kd")
+
+ query = "UPDATE StandardString1 SET :c1 = :v1, :c2 = :v2 WHERE KEY = :key"
+ dbconn.execute(query, dict(c1="ca1", v1="va1", c2="col", v2="val", key="ka"))
+ dbconn.execute(query, dict(c1="cb1", v1="vb1", c2="col", v2="val", key="kb"))
+ dbconn.execute(query, dict(c1="cc1", v1="vc1", c2="col", v2="val", key="kc"))
+ dbconn.execute(query, dict(c1="cd1", v1="vd1", c2="col", v2="val", key="kd"))
dbconn.execute("""
BEGIN BATCH USING CONSISTENCY ONE
@@ -90,7 +90,7 @@ def load_sample(dbconn):
UPDATE StandardLongA SET 5='5', 6='6', 7='8', 9='9' WHERE KEY='ag'
APPLY BATCH
""")
-
+
dbconn.execute("""
BEGIN BATCH USING CONSISTENCY ONE
UPDATE StandardIntegerA SET 10='a', 20='b', 30='c', 40='d' WHERE KEY='k1';
@@ -114,255 +114,294 @@ def load_sample(dbconn):
""")
def init(keyspace="Keyspace1"):
- dbconn = Connection('localhost', 9170, keyspace)
- load_sample(dbconn)
- return dbconn
+ dbconn = cql.connect('localhost', 9170, keyspace)
+ cursor = dbconn.cursor()
+ load_sample(cursor)
+ return cursor
class TestCql(ThriftTester):
def test_select_simple(self):
"retrieve a column"
- conn = init()
- r = conn.execute("SELECT 'ca1' FROM StandardString1 WHERE KEY='ka'")
- assert r[0].key == 'ka'
- assert r[0].columns[0].name == 'ca1'
- assert r[0].columns[0].value == 'va1'
+ cursor = init()
+ cursor.execute("SELECT 'ca1' FROM StandardString1 WHERE KEY='ka'")
+ r = cursor.fetchone()
+ d = cursor.description
+
+ assert d[0][0] == cql.ROW_KEY
+ assert r[0] == 'ka'
+
+ assert d[1][0] == 'ca1'
+ assert r[1] == 'va1'
def test_select_columns(self):
"retrieve multiple columns"
- conn = init()
- r = conn.execute("""
+ cursor = init()
+ cursor.execute("""
SELECT 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
""")
- assert "cd1" in [i.name for i in r[0].columns]
- assert "col" in [i.name for i in r[0].columns]
+
+ d = cursor.description
+ assert "cd1" in [col_dscptn[0] for col_dscptn in d]
+ assert "col" in [col_dscptn[0] for col_dscptn in d]
def test_select_row_range(self):
"retrieve a range of rows with columns"
- conn = init()
- r = conn.execute("""
+ cursor = init()
+ cursor.execute("""
SELECT 4 FROM StandardLongA WHERE KEY > 'ad' AND KEY < 'ag';
""")
- assert len(r) == 4
- assert r[0].key == "ad"
- assert r[1].key == "ae"
- assert r[2].key == "af"
- assert len(r[0].columns) == 1
- assert len(r[1].columns) == 1
- assert len(r[2].columns) == 1
+ keys = ['ad', 'ae', 'af']
+ assert cursor.rowcount == 4
+ for i in range(3):
+ r = cursor.fetchone()
+ assert len(r) == 2
+ assert r[0] == keys[i]
def test_select_row_range_with_limit(self):
"retrieve a limited range of rows with columns"
- conn = init()
- r = conn.execute("""
+ cursor = init()
+ cursor.execute("""
SELECT 1,5,9 FROM StandardLongA WHERE KEY > 'aa'
AND KEY < 'ag' LIMIT 3
""")
- assert len(r) == 3
-
- r = conn.execute("""
+ assert cursor.rowcount == 3
+
+ cursor.execute("""
SELECT 20,40 FROM StandardIntegerA WHERE KEY > 'k1'
AND KEY < 'k7' LIMIT 5
""")
- assert len(r) == 5
- r[0].key == "k1"
- r[4].key == "k5"
+ assert cursor.rowcount == 5
+ for i in range(5):
+ r = cursor.fetchone()
+ assert r[0] == "k%d" % (i+1)
def test_select_columns_slice(self):
"range of columns (slice) by row"
- conn = init()
- r = conn.execute("SELECT 1..3 FROM StandardLongA WHERE KEY = 'aa';")
- assert len(r) == 1
- assert r[0].columns[0].value == "1"
- assert r[0].columns[1].value == "2"
- assert r[0].columns[2].value == "3"
-
- r = conn.execute("SELECT 10..30 FROM StandardIntegerA WHERE KEY='k1'")
- assert len(r) == 1
- assert r[0].columns[0].value == "a"
- assert r[0].columns[1].value == "b"
- assert r[0].columns[2].value == "c"
-
+ cursor = init()
+
+ cursor.execute("SELECT 1..3 FROM StandardLongA WHERE KEY = 'aa';")
+ assert cursor.rowcount == 1
+ r = cursor.fetchone()
+ assert r[0] == "aa"
+ assert r[1] == "1"
+ assert r[2] == "2"
+ assert r[3] == "3"
+
+ cursor.execute("SELECT 10..30 FROM StandardIntegerA WHERE KEY='k1'")
+ assert cursor.rowcount == 1
+ r = cursor.fetchone()
+ assert r[0] == "k1"
+ assert r[1] == "a"
+ assert r[2] == "b"
+ assert r[3] == "c"
+
def test_select_columns_slice_all(self):
"slice all columns in a row"
- conn = init()
- r = conn.execute("SELECT * FROM StandardString1 WHERE KEY = 'ka';")
- assert len(r[0].columns) == 2
- r = conn.execute("SELECT ''..'' FROM StandardString1 WHERE KEY = 'ka';")
- assert len(r[0].columns) == 2
+ cursor = init()
+ cursor.execute("SELECT * FROM StandardString1 WHERE KEY = 'ka';")
+ r = cursor.fetchone()
+ assert len(r) == 3
+ cursor.execute("SELECT ''..'' FROM StandardString1 WHERE KEY = 'ka';")
+ r = cursor.fetchone()
+ assert len(r) == 3
def test_select_columns_slice_with_limit(self):
"range of columns (slice) by row with limit"
- conn = init()
- r = conn.execute("""
+ cursor = init()
+ cursor.execute("""
SELECT FIRST 1 1..3 FROM StandardLongA WHERE KEY = 'aa';
""")
- assert len(r) == 1
- assert len(r[0].columns) == 1
- assert r[0].columns[0].value == "1"
+ assert cursor.rowcount == 1
+ r = cursor.fetchone()
+ assert len(r) == 2
+ assert r[0] == "aa"
+ assert r[1] == "1"
def test_select_columns_slice_reversed(self):
"range of columns (slice) by row reversed"
- conn = init()
- r = conn.execute("""
+ cursor= init()
+ cursor.execute("""
SELECT FIRST 2 REVERSED 3..1 FROM StandardLongA WHERE KEY = 'aa';
""")
- assert len(r) == 1, "%d != 1" % len(r)
- assert len(r[0].columns) == 2
- assert r[0].columns[0].value == "3"
- assert r[0].columns[1].value == "2"
+ assert cursor.rowcount == 1, "%d != 1" % cursor.rowcount
+ r = cursor.fetchone()
+ assert len(r) == 3
+ assert r[0] == 'aa'
+ assert r[1] == "3"
+ assert r[2] == "2"
def test_error_on_multiple_key_by(self):
"ensure multiple key-bys in where clause excepts"
- conn = init()
- assert_raises(CQLException, conn.execute, """
+ cursor = init()
+ assert_raises(cql.ProgrammingError, cursor.execute, """
SELECT 'col' FROM StandardString1 WHERE KEY = 'ka' AND KEY = 'kb';
""")
def test_index_scan_equality(self):
"indexed scan where column equals value"
- conn = init()
- r = conn.execute("""
+ cursor = init()
+ cursor.execute("""
SELECT 'birthdate' FROM IndexedA WHERE 'birthdate' = 100
""")
+ assert cursor.rowcount == 2
+
+ r = cursor.fetchone()
+ assert r[0] == "asmith"
+ assert len(r) == 2
+
+ r = cursor.fetchone()
+ assert r[0] == "dozer"
assert len(r) == 2
- assert r[0].key == "asmith"
- assert r[1].key == "dozer"
- assert len(r[0].columns) == 1
- assert len(r[1].columns) == 1
def test_index_scan_greater_than(self):
"indexed scan where a column is greater than a value"
- conn = init()
- r = conn.execute("""
+ cursor = init()
+ cursor.execute("""
SELECT 'birthdate' FROM IndexedA WHERE 'birthdate' = 100
AND 'unindexed' > 200
""")
- assert len(r) == 1
- assert r[0].key == "asmith"
+ assert cursor.rowcount == 1
+ r = cursor.fetchone()
+ assert r[0] == "asmith"
def test_index_scan_with_start_key(self):
"indexed scan with a starting key"
- conn = init()
- r = conn.execute("""
+ cursor = init()
+ cursor.execute("""
SELECT 'birthdate' FROM IndexedA WHERE 'birthdate' = 100
AND KEY > 'asmithZ'
""")
- assert len(r) == 1
- assert r[0].key == "dozer"
+ assert cursor.rowcount == 1
+ r = cursor.fetchone()
+ assert r[0] == "dozer"
def test_no_where_clause(self):
"empty where clause (range query w/o start key)"
- conn = init()
- r = conn.execute("SELECT 'col' FROM StandardString1 LIMIT 3")
- assert len(r) == 3
- assert r[0].key == "ka"
- assert r[1].key == "kb"
- assert r[2].key == "kc"
+ cursor = init()
+ cursor.execute("SELECT 'col' FROM StandardString1 LIMIT 3")
+ assert cursor.rowcount == 3
+ rows = cursor.fetchmany(3)
+ assert rows[0][0] == "ka"
+ assert rows[1][0] == "kb"
+ assert rows[2][0] == "kc"
def test_column_count(self):
"getting a result count instead of results"
- conn = init()
- r = conn.execute("""
+ cursor = init()
+ cursor.execute("""
SELECT COUNT(1..4) FROM StandardLongA WHERE KEY = 'aa';
""")
- assert r == 4, "expected 4 results, got %d" % (r and r or 0)
+ r = cursor.fetchone()
+ assert r[0] == 4, "expected 4 results, got %d" % (r and r or 0)
def test_truncate_columnfamily(self):
"truncating a column family"
- conn = init()
- conn.execute('TRUNCATE StandardString1;')
- r = conn.execute("SELECT 'cd1' FROM StandardString1 WHERE KEY = 'kd'")
- assert len(r) == 0
+ cursor = init()
+ cursor.execute('TRUNCATE StandardString1;')
+ cursor.execute("SELECT 'cd1' FROM StandardString1 WHERE KEY = 'kd'")
+ assert cursor.rowcount == 0
def test_delete_columns(self):
"delete columns from a row"
- conn = init()
- r = conn.execute("""
+ cursor = init()
+ cursor.execute("""
SELECT 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
""")
- assert "cd1" in [i.name for i in r[0].columns]
- assert "col" in [i.name for i in r[0].columns]
- conn.execute("""
+ colnames = [col_d[0] for col_d in cursor.description]
+ assert "cd1" in colnames
+ assert "col" in colnames
+ cursor.execute("""
DELETE 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
""")
- r = conn.execute("""
+ cursor.execute("""
SELECT 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
""")
- assert len(r[0].columns) == 0
+ r = cursor.fetchone()
+ assert len(r) == 1
def test_delete_columns_multi_rows(self):
"delete columns from multiple rows"
- conn = init()
- r = conn.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kc'")
- assert len(r[0].columns) == 1
- r = conn.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kd'")
- assert len(r[0].columns) == 1
+ cursor = init()
+
+ cursor.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kc'")
+ r = cursor.fetchone()
+ assert len(r) == 2
+
+ cursor.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kd'")
+ r = cursor.fetchone()
+ assert len(r) == 2
- conn.execute("""
+ cursor.execute("""
DELETE 'col' FROM StandardString1 WHERE KEY IN ('kc', 'kd')
""")
- r = conn.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kc'")
- assert len(r[0].columns) == 0
- r = conn.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kd'")
- assert len(r[0].columns) == 0
+ cursor.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kc'")
+ r = cursor.fetchone()
+ assert len(r) == 1
+
+ cursor.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kd'")
+ r = cursor.fetchone()
+ assert len(r) == 1
def test_delete_rows(self):
"delete entire rows"
- conn = init()
- r = conn.execute("""
+ cursor = init()
+ cursor.execute("""
SELECT 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
""")
- assert "cd1" in [i.name for i in r[0].columns]
- assert "col" in [i.name for i in r[0].columns]
- conn.execute("DELETE FROM StandardString1 WHERE KEY = 'kd'")
- r = conn.execute("""
+ colnames = [col_d[0] for col_d in cursor.description]
+ assert "cd1" in colnames
+ assert "col" in colnames
+ cursor.execute("DELETE FROM StandardString1 WHERE KEY = 'kd'")
+ cursor.execute("""
SELECT 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
""")
- assert len(r[0].columns) == 0
-
+ r = cursor.fetchone()
+ assert len(r) == 1
+
def test_create_keyspace(self):
"create a new keyspace"
- init().execute("""
+ cursor = init()
+ cursor.execute("""
CREATE KEYSPACE TestKeyspace42 WITH strategy_options:DC1 = '1'
AND strategy_class = 'NetworkTopologyStrategy'
""")
-
+
# TODO: temporary (until this can be done with CQL).
ksdef = thrift_client.describe_keyspace("TestKeyspace42")
-
+
strategy_class = "org.apache.cassandra.locator.NetworkTopologyStrategy"
assert ksdef.strategy_class == strategy_class
assert ksdef.strategy_options['DC1'] == "1"
-
+
def test_drop_keyspace(self):
"removing a keyspace"
- conn = init()
- conn.execute("""
- CREATE KEYSPACE Keyspace4Drop
- WITH strategy_class = SimpleStrategy AND strategy_options:replication_factor = 1
+ cursor = init()
+ cursor.execute("""
+ CREATE KEYSPACE Keyspace4Drop WITH strategy_options:replication_factor = '1'
+ AND strategy_class = 'SimpleStrategy'
""")
-
+
# TODO: temporary (until this can be done with CQL).
thrift_client.describe_keyspace("Keyspace4Drop")
-
- conn.execute('DROP KEYSPACE Keyspace4Drop;')
-
+
+ cursor.execute('DROP KEYSPACE Keyspace4Drop;')
+
# Technically this should throw a ttypes.NotFound(), but this is
# temporary and so not worth requiring it on PYTHONPATH.
assert_raises(Exception,
thrift_client.describe_keyspace,
"Keyspace4Drop")
-
+
def test_create_column_family(self):
"create a new column family"
- conn = init()
- conn.execute("""
- CREATE KEYSPACE CreateCFKeyspace WITH strategy_options:replication_factor = 1
- AND strategy_class = 'SimpleStrategy';
- """)
- conn.execute("USE CreateCFKeyspace;")
-
- conn.execute("""
+ cursor = init()
+ cursor.execute("""
+ CREATE KEYSPACE CreateCFKeyspace WITH strategy_options:replication_factor = '1'
+ AND strategy_class = 'SimpleStrategy';
+ """)
+ cursor.execute("USE CreateCFKeyspace;")
+
+ cursor.execute("""
CREATE COLUMNFAMILY NewCf1 (
KEY varint PRIMARY KEY,
'username' text,
@@ -371,7 +410,7 @@ class TestCql(ThriftTester):
'id' uuid
) WITH comment = 'shiny, new, cf' AND default_validation = ascii;
""")
-
+
# TODO: temporary (until this can be done with CQL).
ksdef = thrift_client.describe_keyspace("CreateCFKeyspace")
assert len(ksdef.cf_defs) == 1, \
@@ -383,27 +422,27 @@ class TestCql(ThriftTester):
assert cfam.default_validation_class == "org.apache.cassandra.db.marshal.AsciiType"
assert cfam.comparator_type == "org.apache.cassandra.db.marshal.UTF8Type"
assert cfam.key_validation_class == "org.apache.cassandra.db.marshal.IntegerType"
-
+
# Missing primary key
- assert_raises(CQLException, conn.execute, "CREATE COLUMNFAMILY NewCf2")
-
+ assert_raises(cql.ProgrammingError, cursor.execute, "CREATE COLUMNFAMILY NewCf2")
+
# Too many primary keys
- assert_raises(CQLException,
- conn.execute,
+ assert_raises(cql.ProgrammingError,
+ cursor.execute,
"""CREATE COLUMNFAMILY NewCf2
(KEY varint PRIMARY KEY, KEY text PRIMARY KEY)""")
-
+
# No column defs
- conn.execute("""CREATE COLUMNFAMILY NewCf3
+ cursor.execute("""CREATE COLUMNFAMILY NewCf3
(KEY varint PRIMARY KEY) WITH comparator = bigint""")
ksdef = thrift_client.describe_keyspace("CreateCFKeyspace")
assert len(ksdef.cf_defs) == 2, \
"expected 3 column families total, found %d" % len(ksdef.cf_defs)
cfam = [i for i in ksdef.cf_defs if i.name == "NewCf3"][0]
assert cfam.comparator_type == "org.apache.cassandra.db.marshal.LongType"
-
+
# Column defs, defaults otherwise
- conn.execute("""CREATE COLUMNFAMILY NewCf4
+ cursor.execute("""CREATE COLUMNFAMILY NewCf4
(KEY varint PRIMARY KEY, 'a' varint, 'b' varint)
WITH comparator = text;""")
ksdef = thrift_client.describe_keyspace("CreateCFKeyspace")
@@ -415,34 +454,34 @@ class TestCql(ThriftTester):
for coldef in cfam.column_metadata:
assert coldef.name in ("a", "b"), "Unknown column name " + coldef.name
assert coldef.validation_class.endswith("marshal.IntegerType")
-
+
def test_drop_columnfamily(self):
"removing a column family"
- conn = init()
- conn.execute("""
- CREATE KEYSPACE Keyspace4CFDrop WITH strategy_options:replication_factor = 1
- AND strategy_class = 'SimpleStrategy';
- """)
- conn.execute('USE Keyspace4CFDrop;')
- conn.execute('CREATE COLUMNFAMILY CF4Drop (KEY varint PRIMARY KEY);')
-
+ cursor = init()
+ cursor.execute("""
+ CREATE KEYSPACE Keyspace4CFDrop WITH strategy_options:replication_factor = '1'
+ AND strategy_class = 'SimpleStrategy';
+ """)
+ cursor.execute('USE Keyspace4CFDrop;')
+ cursor.execute('CREATE COLUMNFAMILY CF4Drop (KEY varint PRIMARY KEY);')
+
# TODO: temporary (until this can be done with CQL).
ksdef = thrift_client.describe_keyspace("Keyspace4CFDrop")
assert len(ksdef.cf_defs), "Column family not created!"
-
- conn.execute('DROP COLUMNFAMILY CF4Drop;')
-
+
+ cursor.execute('DROP COLUMNFAMILY CF4Drop;')
+
ksdef = thrift_client.describe_keyspace("Keyspace4CFDrop")
assert not len(ksdef.cf_defs), "Column family not deleted!"
-
+
def test_create_indexs(self):
"creating column indexes"
- conn = init()
- conn.execute("USE Keyspace1")
- conn.execute("CREATE COLUMNFAMILY CreateIndex1 (KEY text PRIMARY KEY)")
- conn.execute("CREATE INDEX namedIndex ON CreateIndex1 (items)")
- conn.execute("CREATE INDEX ON CreateIndex1 (stuff)")
-
+ cursor = init()
+ cursor.execute("USE Keyspace1")
+ cursor.execute("CREATE COLUMNFAMILY CreateIndex1 (KEY text PRIMARY KEY)")
+ cursor.execute("CREATE INDEX namedIndex ON CreateIndex1 (items)")
+ cursor.execute("CREATE INDEX ON CreateIndex1 (stuff)")
+
# TODO: temporary (until this can be done with CQL).
ksdef = thrift_client.describe_keyspace("Keyspace1")
cfam = [i for i in ksdef.cf_defs if i.name == "CreateIndex1"][0]
@@ -455,205 +494,208 @@ class TestCql(ThriftTester):
assert stuff.index_type == 0, "missing index"
# already indexed
- assert_raises(CQLException,
- conn.execute,
+ assert_raises(cql.ProgrammingError,
+ cursor.execute,
"CREATE INDEX ON CreateIndex1 (stuff)")
def test_time_uuid(self):
"store and retrieve time-based (type 1) uuids"
- conn = init()
-
+ cursor = init()
+
# Store and retrieve a timeuuid using it's hex-formatted string
timeuuid = uuid.uuid1()
- conn.execute("""
+ cursor.execute("""
UPDATE StandardTimeUUID SET '%s' = 10 WHERE KEY = 'uuidtest'
""" % str(timeuuid))
-
- r = conn.execute("""
+
+ cursor.execute("""
SELECT '%s' FROM StandardTimeUUID WHERE KEY = 'uuidtest'
""" % str(timeuuid))
- assert r[0].columns[0].name == timeuuid
-
+ d = cursor.description
+ assert d[1][0] == timeuuid, "%s, %s" % (str(d[1][0]), str(timeuuid))
+
# Tests a node-side conversion from bigint to UUID.
ms = uuid1bytes_to_millis(uuid.uuid1().bytes)
- conn.execute("""
+ cursor.execute("""
UPDATE StandardTimeUUIDValues SET 'id' = %d WHERE KEY = 'uuidtest'
""" % ms)
-
- r = conn.execute("""
+
+ cursor.execute("""
SELECT 'id' FROM StandardTimeUUIDValues WHERE KEY = 'uuidtest'
""")
- assert uuid1bytes_to_millis(r[0].columns[0].value.bytes) == ms
-
+ r = cursor.fetchone()
+ assert uuid1bytes_to_millis(r[1].bytes) == ms
+
# Tests a node-side conversion from ISO8601 to UUID.
- conn.execute("""
+ cursor.execute("""
UPDATE StandardTimeUUIDValues SET 'id2' = '2011-01-31 17:00:00-0000'
WHERE KEY = 'uuidtest'
""")
-
- r = conn.execute("""
+
+ cursor.execute("""
SELECT 'id2' FROM StandardTimeUUIDValues WHERE KEY = 'uuidtest'
""")
# 2011-01-31 17:00:00-0000 == 1296493200000ms
- ms = uuid1bytes_to_millis(r[0].columns[0].value.bytes)
+ r = cursor.fetchone()
+ ms = uuid1bytes_to_millis(r[1].bytes)
assert ms == 1296493200000, \
"%d != 1296493200000 (2011-01-31 17:00:00-0000)" % ms
# Tests node-side conversion of timeuuid("now") to UUID
- conn.execute("""
+ cursor.execute("""
UPDATE StandardTimeUUIDValues SET 'id3' = 'now'
WHERE KEY = 'uuidtest'
""")
-
- r = conn.execute("""
+
+ cursor.execute("""
SELECT 'id3' FROM StandardTimeUUIDValues WHERE KEY = 'uuidtest'
""")
- ms = uuid1bytes_to_millis(r[0].columns[0].value.bytes)
+ r = cursor.fetchone()
+ ms = uuid1bytes_to_millis(r[1].bytes)
assert ((time.time() * 1e3) - ms) < 100, \
"new timeuuid not within 100ms of now (UPDATE vs. SELECT)"
uuid_range = []
- update = "UPDATE StandardTimeUUID SET ? = ? WHERE KEY = slicetest"
+ update = "UPDATE StandardTimeUUID SET :name = :val WHERE KEY = slicetest"
for i in range(5):
uuid_range.append(uuid.uuid1())
- conn.execute(update, uuid_range[i], i)
+ cursor.execute(update, dict(name=uuid_range[i], val=i))
+
+ cursor.execute("""
+ SELECT :start..:finish FROM StandardTimeUUID WHERE KEY = slicetest
+ """, dict(start=uuid_range[0], finish=uuid_range[len(uuid_range)-1]))
+ d = cursor.description
+ for (i, col_d) in enumerate(d[1:]):
+ assert uuid_range[i] == col_d[0]
+
- r = conn.execute("""
- SELECT ?..? FROM StandardTimeUUID WHERE KEY = slicetest
- """, uuid_range[0], uuid_range[len(uuid_range)-1])
-
- for (i, col) in enumerate(r[0]):
- assert uuid_range[i] == col.name
-
-
def test_lexical_uuid(self):
"store and retrieve lexical uuids"
- conn = init()
+ cursor = init()
uid = uuid.uuid4()
- conn.execute("UPDATE StandardUUID SET ? = 10 WHERE KEY = 'uuidtest'",
- uid)
-
- r = conn.execute("SELECT ? FROM StandardUUID WHERE KEY = 'uuidtest'",
- uid)
- assert r[0].columns[0].name == uid, r[0].columns[0].name
-
+ cursor.execute("UPDATE StandardUUID SET :name = 10 WHERE KEY = 'uuidtest'",
+ dict(name=uid))
+
+ cursor.execute("SELECT :name FROM StandardUUID WHERE KEY = 'uuidtest'",
+ dict(name=uid))
+ d = cursor.description
+ assert d[1][0] == uid, "expected %s, got %s (%s)" % \
+ (uid.bytes.encode('hex'), str(d[1][0]).encode('hex'), d[1][1])
+
# TODO: slices of uuids from cf w/ LexicalUUIDType comparator
-
+
def test_utf8_read_write(self):
"reading and writing utf8 values"
- conn = init()
+ cursor = init()
# Sorting: ¢ (u00a2) < © (u00a9) < ® (u00ae) < ¿ (u00bf)
- conn.execute("UPDATE StandardUtf82 SET ? = v1 WHERE KEY = k1", "¿")
- conn.execute("UPDATE StandardUtf82 SET ? = v1 WHERE KEY = k1", "©")
- conn.execute("UPDATE StandardUtf82 SET ? = v1 WHERE KEY = k1", "®")
- conn.execute("UPDATE StandardUtf82 SET ? = v1 WHERE KEY = k1", "¢")
-
- r = conn.execute("SELECT * FROM StandardUtf82 WHERE KEY = k1")
- assert r[0][0].name == u"¢"
- assert r[0][1].name == u"©"
- assert r[0][2].name == u"®"
- assert r[0][3].name == u"¿"
-
- r = conn.execute("SELECT ?..'' FROM StandardUtf82 WHERE KEY = k1", "©")
- assert len(r[0]) == 3
- assert r[0][0].name == u"©"
- assert r[0][1].name == u"®"
- assert r[0][2].name == u"¿"
-
+ cursor.execute("UPDATE StandardUtf82 SET :name = v1 WHERE KEY = k1", dict(name="¿"))
+ cursor.execute("UPDATE StandardUtf82 SET :name = v1 WHERE KEY = k1", dict(name="©"))
+ cursor.execute("UPDATE StandardUtf82 SET :name = v1 WHERE KEY = k1", dict(name="®"))
+ cursor.execute("UPDATE StandardUtf82 SET :name = v1 WHERE KEY = k1", dict(name="¢"))
+
+ cursor.execute("SELECT * FROM StandardUtf82 WHERE KEY = k1")
+ d = cursor.description
+ assert d[1][0] == u"¢", d[1][0]
+ assert d[2][0] == u"©", d[2][0]
+ assert d[3][0] == u"®", d[3][0]
+ assert d[4][0] == u"¿", d[4][0]
+
+ cursor.execute("SELECT :start..'' FROM StandardUtf82 WHERE KEY = k1", dict(start="©"))
+ r = cursor.fetchone()
+ assert len(r) == 4
+ d = cursor.description
+ assert d[1][0] == u"©"
+ assert d[2][0] == u"®"
+ assert d[3][0] == u"¿"
+
def test_read_write_negative_numerics(self):
"reading and writing negative numeric values"
- conn = init()
+ cursor = init()
for cf in ("StandardIntegerA", "StandardLongA"):
for i in range(10):
- conn.execute("UPDATE ? SET ? = ? WHERE KEY = negatives;",
- cf,
- -(i + 1),
- i)
- r = conn.execute("SELECT ?..? FROM ? WHERE KEY = negatives;",
- -10,
- -1,
- cf)
- assert len(r[0]) == 10, \
- "returned %d columns, expected %d" % (len(r[0]), 10)
- assert r[0][0].name == -10
- assert r[0][9].name == -1
-
+ cursor.execute("UPDATE :cf SET :name = :val WHERE KEY = negatives;",
+ dict(cf=cf, name=-(i + 1), val=i))
+
+ cursor.execute("SELECT :start..:finish FROM :cf WHERE KEY = negatives;",
+ dict(start=-10, finish=-1, cf=cf))
+ r = cursor.fetchone()
+ assert len(r) == 11, \
+ "returned %d columns, expected %d" % (len(r) - 1, 10)
+ d = cursor.description
+ assert d[1][0] == -10
+ assert d[10][0] == -1
+
def test_escaped_quotes(self):
"reading and writing strings w/ escaped quotes"
- conn = init()
-
- conn.execute("""
- UPDATE StandardString1 SET 'x''and''y' = z WHERE KEY = ?
- """, "test_escaped_quotes")
-
- r = conn.execute("""
- SELECT 'x''and''y' FROM StandardString1 WHERE KEY = ?
- """, "test_escaped_quotes")
- assert (len(r) == 1) and (len(r[0]) == 1), "wrong number of results"
- assert r[0][0].name == "x\'and\'y"
-
+ cursor = init()
+
+ cursor.execute("""
+ UPDATE StandardString1 SET 'x''and''y' = z WHERE KEY = :key
+ """, dict(key="test_escaped_quotes"))
+
+ cursor.execute("""
+ SELECT 'x''and''y' FROM StandardString1 WHERE KEY = :key
+ """, dict(key="test_escaped_quotes"))
+ assert cursor.rowcount == 1
+ r = cursor.fetchone()
+ assert len(r) == 2, "wrong number of results"
+ d = cursor.description
+ assert d[1][0] == "x\'and\'y"
+
def test_typed_keys(self):
"using typed keys"
- conn = init()
- r = conn.execute("SELECT * FROM StandardString1 WHERE KEY = ?", "ka")
- assert isinstance(r[0].key, unicode), \
- "wrong key-type returned, expected unicode, got %s" % type(r[0].key)
-
+ cursor = init()
+ cursor.execute("SELECT * FROM StandardString1 WHERE KEY = :key", dict(key="ka"))
+ r = cursor.fetchone()
+ assert isinstance(r[0], unicode), \
+ "wrong key-type returned, expected unicode, got %s" % type(r[0])
+
# FIXME: The above is woefully inadequate, but the test config uses
# CollatingOrderPreservingPartitioner which only supports UTF8.
-
+
def test_write_using_insert(self):
"peforming writes using \"insert\""
- conn = init()
- conn.execute("INSERT INTO StandardUtf82 (KEY, ?, ?) VALUES (?, ?, ?)",
- "pork",
- "beef",
- "meat",
- "bacon",
- "brisket")
-
- r = conn.execute("SELECT * FROM StandardUtf82 WHERE KEY = ?", "meat")
- assert r[0][0].name == "beef"
- assert r[0][0].value == "brisket"
- assert r[0][1].name == "pork"
- assert r[0][1].value == "bacon"
-
+ cursor = init()
+ cursor.execute("INSERT INTO StandardUtf82 (KEY, :c1, :c2) VALUES (:key, :v1, :v2)",
+ dict(c1="pork", c2="beef", key="meat", v1="bacon", v2="brisket"))
+
+ cursor.execute("SELECT * FROM StandardUtf82 WHERE KEY = :key", dict(key="meat"))
+ r = cursor.fetchone()
+ d = cursor.description
+ assert d[1][0] == "beef"
+ assert r[1] == "brisket"
+
+ assert d[2][0] == "pork"
+ assert r[2] == "bacon"
+
# Bad writes.
-
+
# Too many column values
- assert_raises(CQLException,
- conn.execute,
- "INSERT INTO StandardUtf82 (KEY, ?) VALUES (?, ?, ?)",
- "name1",
- "key0",
- "value1",
- "value2")
-
+ assert_raises(cql.ProgrammingError,
+ cursor.execute,
+ "INSERT INTO StandardUtf82 (KEY, :c1) VALUES (:key, :v1, :v2)",
+ dict(c1="name1", key="key0", v1="value1", v2="value2"))
+
# Too many column names, (not enough column values)
- assert_raises(CQLException,
- conn.execute,
- "INSERT INTO StandardUtf82 (KEY, ?, ?) VALUES (?, ?)",
- "name1",
- "name2",
- "key0",
- "value1")
-
+ assert_raises(cql.ProgrammingError,
+ cursor.execute,
+ "INSERT INTO StandardUtf82 (KEY, :c1, :c2) VALUES (:key, :v1)",
+ dict(c1="name1", c2="name2", key="key0", v1="value1"))
+
def test_compression_disabled(self):
"reading and writing w/ compression disabled"
- conn = init()
- conn.execute("UPDATE StandardString1 SET ? = ? WHERE KEY = ?",
- "some_name",
- "some_value",
- "compression_test",
- compression='NONE')
-
- r = conn.execute("SELECT ? FROM StandardString1 WHERE KEY = ?",
- "some_name",
- "compression_test",
- compression='NONE')
-
- assert len(r) == 1, "expected 1 result, got %d" % len(r)
- assert r[0][0].name == "some_name", \
- "unrecognized name '%s'" % r[0][0].name
- assert r[0][0].value == "some_value", \
- "unrecognized value '%s'" % r[0][0].value
+ cursor = init()
+ cursor.compression = 'NONE'
+ cursor.execute("UPDATE StandardString1 SET :name = :val WHERE KEY = :key",
+ dict(name="some_name", val="some_value", key="compression_test"))
+
+ cursor.execute("SELECT :name FROM StandardString1 WHERE KEY = :key",
+ dict(name="some_name", key="compression_test"))
+
+ assert cursor.rowcount == 1, "expected 1 result, got %d" % cursor.rowcount
+ colnames = [col_d[0] for col_d in cursor.description]
+ assert colnames[1] == "some_name", \
+ "unrecognized name '%s'" % colnames[1]
+ r = cursor.fetchone()
+ assert r[1] == "some_value", \
+ "unrecognized value '%s'" % r[1]