From: Mike Bayer Date: Fri, 4 May 2012 21:43:21 +0000 (-0400) Subject: - [feature] Added SQLite execution option X-Git-Tag: rel_0_7_7~8 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=c43ac1dc1c98353a438bc783f62a4af1093073cc;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git - [feature] Added SQLite execution option "sqlite_raw_colnames=True", will bypass attempts to remove "." from column names returned by SQLite cursor.description. [ticket:2475] --- diff --git a/CHANGES b/CHANGES index ef1965d18e..0a01dfd132 100644 --- a/CHANGES +++ b/CHANGES @@ -123,6 +123,15 @@ CHANGES grouped; grouping is now applied to those expressions. [ticket:2467] +- sqlite + + - [feature] Added SQLite execution option + "sqlite_raw_colnames=True", will bypass + attempts to remove "." from column names + returned by SQLite cursor.description. + [ticket:2475] + + 0.7.6 ===== - orm diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py index b9cd783bef..e309d20fe8 100644 --- a/lib/sqlalchemy/dialects/sqlite/base.py +++ b/lib/sqlalchemy/dialects/sqlite/base.py @@ -441,6 +441,22 @@ class SQLiteIdentifierPreparer(compiler.IdentifierPreparer): result = self.quote_schema(index.table.schema, index.table.quote_schema) + "." + result return result +class SQLiteExecutionContext(default.DefaultExecutionContext): + @util.memoized_property + def _preserve_raw_colnames(self): + return self.execution_options.get("sqlite_raw_colnames", False) + + def _translate_colname(self, colname): + # adjust for dotted column names. SQLite + # in the case of UNION may store col names as + # "tablename.colname" + # in cursor.description + if not self._preserve_raw_colnames and "." in colname: + return colname.split(".")[1], colname + else: + return colname, None + + class SQLiteDialect(default.DefaultDialect): name = 'sqlite' supports_alter = False @@ -451,6 +467,7 @@ class SQLiteDialect(default.DefaultDialect): supports_cast = True default_paramstyle = 'qmark' + execution_ctx_cls = SQLiteExecutionContext statement_compiler = SQLiteCompiler ddl_compiler = SQLiteDDLCompiler type_compiler = SQLiteTypeCompiler @@ -525,16 +542,6 @@ class SQLiteDialect(default.DefaultDialect): else: return None - def _translate_colname(self, colname): - # adjust for dotted column names. SQLite - # in the case of UNION may store col names as - # "tablename.colname" - # in cursor.description - if "." in colname: - return colname.split(".")[1], colname - else: - return colname, None - @reflection.cache def get_table_names(self, connection, schema=None, **kw): if schema is not None: diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py index 1d25113337..6d18db16c7 100644 --- a/lib/sqlalchemy/engine/base.py +++ b/lib/sqlalchemy/engine/base.py @@ -2723,7 +2723,7 @@ class ResultMetaData(object): context = parent.context dialect = context.dialect typemap = dialect.dbapi_type_map - translate_colname = dialect._translate_colname + translate_colname = context._translate_colname # high precedence key values. primary_keymap = {} diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py index d0cbe871ff..f3d05cac27 100644 --- a/lib/sqlalchemy/engine/default.py +++ b/lib/sqlalchemy/engine/default.py @@ -96,10 +96,6 @@ class DefaultDialect(base.Dialect): # and denormalize_name() must be provided. requires_name_normalize = False - # a hook for SQLite's translation of - # result column names - _translate_colname = None - reflection_options = () def __init__(self, convert_unicode=False, assert_unicode=False, @@ -358,6 +354,10 @@ class DefaultExecutionContext(base.ExecutionContext): _is_implicit_returning = False _is_explicit_returning = False + # a hook for SQLite's translation of + # result column names + _translate_colname = None + @classmethod def _init_ddl(cls, dialect, connection, dbapi_connection, compiled_ddl): """Initialize execution context for a DDLElement construct.""" diff --git a/test/sql/test_query.py b/test/sql/test_query.py index 9f0c2dab09..ed6f59a548 100644 --- a/test/sql/test_query.py +++ b/test/sql/test_query.py @@ -716,38 +716,91 @@ class QueryTest(fixtures.TestBase): self.assert_(r[1:] == (2, 'foo@bar.com')) self.assert_(r[:-1] == (1, 2)) - def test_column_accessor(self): - users.insert().execute(user_id=1, user_name='john') - users.insert().execute(user_id=2, user_name='jack') - addresses.insert().execute(address_id=1, user_id=2, address='foo@bar.com') + def test_column_accessor_basic_compiled(self): + users.insert().execute( + dict(user_id=1, user_name='john'), + dict(user_id=2, user_name='jack') + ) r = users.select(users.c.user_id==2).execute().first() self.assert_(r.user_id == r['user_id'] == r[users.c.user_id] == 2) self.assert_(r.user_name == r['user_name'] == r[users.c.user_name] == 'jack') + def test_column_accessor_basic_text(self): + users.insert().execute( + dict(user_id=1, user_name='john'), + dict(user_id=2, user_name='jack') + ) r = text("select * from query_users where user_id=2", bind=testing.db).execute().first() self.assert_(r.user_id == r['user_id'] == r[users.c.user_id] == 2) self.assert_(r.user_name == r['user_name'] == r[users.c.user_name] == 'jack') + def test_column_accessor_dotted_union(self): + users.insert().execute( + dict(user_id=1, user_name='john'), + ) + # test a little sqlite weirdness - with the UNION, # cols come back as "query_users.user_id" in cursor.description r = text("select query_users.user_id, query_users.user_name from query_users " "UNION select query_users.user_id, query_users.user_name from query_users", bind=testing.db).execute().first() - self.assert_(r['user_id']) == 1 - self.assert_(r['user_name']) == "john" + eq_(r['user_id'], 1) + eq_(r['user_name'], "john") + eq_(r.keys(), ["user_id", "user_name"]) + @testing.only_on("sqlite", "sqlite specific feature") + def test_column_accessor_sqlite_raw(self): + users.insert().execute( + dict(user_id=1, user_name='john'), + ) + + r = text("select query_users.user_id, query_users.user_name from query_users " + "UNION select query_users.user_id, query_users.user_name from query_users", + bind=testing.db).execution_options(sqlite_raw_colnames=True).execute().first() + assert 'user_id' not in r + assert 'user_name' not in r + eq_(r['query_users.user_id'], 1) + eq_(r['query_users.user_name'], "john") + eq_(r.keys(), ["query_users.user_id", "query_users.user_name"]) + + @testing.only_on("sqlite", "sqlite specific feature") + def test_column_accessor_sqlite_translated(self): + users.insert().execute( + dict(user_id=1, user_name='john'), + ) + + r = text("select query_users.user_id, query_users.user_name from query_users " + "UNION select query_users.user_id, query_users.user_name from query_users", + bind=testing.db).execute().first() + eq_(r['user_id'], 1) + eq_(r['user_name'], "john") + eq_(r['query_users.user_id'], 1) + eq_(r['query_users.user_name'], "john") + eq_(r.keys(), ["user_id", "user_name"]) + + def test_column_accessor_labels_w_dots(self): + users.insert().execute( + dict(user_id=1, user_name='john'), + ) # test using literal tablename.colname r = text('select query_users.user_id AS "query_users.user_id", ' 'query_users.user_name AS "query_users.user_name" from query_users', - bind=testing.db).execute().first() - self.assert_(r['query_users.user_id']) == 1 - self.assert_(r['query_users.user_name']) == "john" + bind=testing.db).execution_options(sqlite_raw_colnames=True).execute().first() + eq_(r['query_users.user_id'], 1) + eq_(r['query_users.user_name'], "john") + assert "user_name" not in r + eq_(r.keys(), ["query_users.user_id", "query_users.user_name"]) + + def test_column_accessor_unary(self): + users.insert().execute( + dict(user_id=1, user_name='john'), + ) # unary experssions r = select([users.c.user_name.distinct()]).order_by(users.c.user_name).execute().first() - eq_(r[users.c.user_name], 'jack') - eq_(r.user_name, 'jack') + eq_(r[users.c.user_name], 'john') + eq_(r.user_name, 'john') def test_column_accessor_err(self): r = testing.db.execute(select([1])).first()