- STRING/FIXED_CHAR now convert to unicode natively.
SQLAlchemy's String types then don't need to
apply any kind of conversions.
-
+
+- firebird
+ - The functionality of result.rowcount is now disabled
+ by default, and can be re-enabled using the 'enable_rowcount'
+ flag with create_engine(), as well as the 'enable_rowcount'
+ execution context flag on a per-execute basis. This because
+ cursor.rowcount requires cursor access (can't be evaluated
+ lazily since the result auto-closes) and also incurs an
+ expensive round-trip.
+
- examples
- Updated attribute_shard.py example to use a more robust
method of searching a Query for binary expressions which
Kinterbasedb backend specific keyword arguments are:
-type_conv
- select the kind of mapping done on the types: by default SQLAlchemy
+* type_conv - select the kind of mapping done on the types: by default SQLAlchemy
uses 200 with Unicode, datetime and decimal support (see details__).
-concurrency_level
- set the backend policy with regards to threading issues: by default
+* concurrency_level - set the backend policy with regards to threading issues: by default
SQLAlchemy uses policy 1 (see details__).
+* enable_rowcount - False by default, this enables the usage of "cursor.rowcount" with the
+ Kinterbasdb dialect. When disabled, SQLAlchemy's ResultProxy will
+ return -1 for result.rowcount. The rationale here is that Kinterbasdb
+ requires a second round trip to the database when .rowcount is called -
+ since SQLA's resultproxy automatically closes the cursor after a
+ non-result-returning statement, rowcount must be called, if at all,
+ before the result object is returned. The behavior can also be
+ controlled on a per-execution basis using the `enable_rowcount`
+ option with :meth:`execution_options()`::
+
+ conn = engine.connect().execution_options(enable_rowcount=True)
+ r = conn.execute(stmt)
+ print r.rowcount
+
__ http://sourceforge.net/projects/kinterbasdb
__ http://firebirdsql.org/index.php?op=devel&sub=python
__ http://kinterbasdb.sourceforge.net/dist_docs/usage.html#adv_param_conv_dynamic_type_translation
__ http://kinterbasdb.sourceforge.net/dist_docs/usage.html#special_issue_concurrency
"""
-from sqlalchemy.dialects.firebird.base import FBDialect, FBCompiler
+from sqlalchemy.dialects.firebird.base import FBDialect, \
+ FBCompiler, FBExecutionContext
from sqlalchemy import util, types as sqltypes
class _FBNumeric_kinterbasdb(sqltypes.Numeric):
else:
return value
return process
-
+
+class FBExecutionContext_kinterbasdb(FBExecutionContext):
+ @property
+ def rowcount(self):
+ if self.execution_options.get('enable_rowcount',
+ self.dialect.enable_rowcount):
+ return self.cursor.rowcount
+ else:
+ return -1
+
class FBDialect_kinterbasdb(FBDialect):
driver = 'kinterbasdb'
supports_sane_rowcount = False
supports_sane_multi_rowcount = False
+ execution_ctx_cls = FBExecutionContext_kinterbasdb
supports_native_decimal = True
)
- def __init__(self, type_conv=200, concurrency_level=1, **kwargs):
+ def __init__(self, type_conv=200, concurrency_level=1, enable_rowcount=False, **kwargs):
super(FBDialect_kinterbasdb, self).__init__(**kwargs)
-
+ self.enable_rowcount = enable_rowcount
self.type_conv = type_conv
self.concurrency_level = concurrency_level
-
+ if enable_rowcount:
+ self.supports_sane_rowcount = True
+
@classmethod
def dbapi(cls):
k = __import__('kinterbasdb')
class MiscTest(TestBase):
__only_on__ = 'firebird'
+ @testing.provide_metadata
def test_strlen(self):
# On FB the length() function is implemented by an external
# UDF, strlen(). Various SA tests fail because they pass a
# the maximum string length the UDF was declared to accept).
# This test checks that at least it works ok in other cases.
- meta = MetaData(testing.db)
- t = Table('t1', meta,
+ t = Table('t1', metadata,
Column('id', Integer, Sequence('t1idseq'), primary_key=True),
Column('name', String(10))
)
- meta.create_all()
- try:
- t.insert(values=dict(name='dante')).execute()
- t.insert(values=dict(name='alighieri')).execute()
- select([func.count(t.c.id)],func.length(t.c.name)==5).execute().first()[0] == 1
- finally:
- meta.drop_all()
+ metadata.create_all()
+ t.insert(values=dict(name='dante')).execute()
+ t.insert(values=dict(name='alighieri')).execute()
+ select([func.count(t.c.id)],func.length(t.c.name)==5).execute().first()[0] == 1
def test_server_version_info(self):
version = testing.db.dialect.server_version_info
assert len(version) == 3, "Got strange version info: %s" % repr(version)
+ @testing.provide_metadata
+ def test_rowcount_flag(self):
+ engine = engines.testing_engine(options={'enable_rowcount':True})
+ assert engine.dialect.supports_sane_rowcount
+ metadata.bind = engine
+ t = Table('t1', metadata,
+ Column('data', String(10))
+ )
+ metadata.create_all()
+ r = t.insert().execute({'data':'d1'}, {'data':'d2'}, {'data': 'd3'})
+ r = t.update().where(t.c.data=='d2').values(data='d3').execute()
+ eq_(r.rowcount, 1)
+ r = t.delete().where(t.c.data == 'd3').execute()
+ eq_(r.rowcount, 2)
+
+ r = t.delete().execution_options(enable_rowcount=False).execute()
+ eq_(r.rowcount, -1)
+
+
+ engine = engines.testing_engine(options={'enable_rowcount':False})
+ assert not engine.dialect.supports_sane_rowcount
+ metadata.bind = engine
+ r = t.insert().execute({'data':'d1'}, {'data':'d2'}, {'data':'d3'})
+ r = t.update().where(t.c.data=='d2').values(data='d3').execute()
+ eq_(r.rowcount, -1)
+ r = t.delete().where(t.c.data == 'd3').execute()
+ eq_(r.rowcount, -1)
+ r = t.delete().execution_options(enable_rowcount=True).execute()
+ eq_(r.rowcount, 1)
+
def test_percents_in_text(self):
for expr, result in (
(text("select '%' from rdb$database"), '%'),
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
