- [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 ef1965d18e6ef87e0cb717ab8925ab653af05e64..0a01dfd132a9810912173478116fde49e026f854 100644 (file)
--- 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 b9cd783beff9eded37ff618097d3f06df6644a26..e309d20fe8b47310adf81de22a587e356d9a6ee9 100644 (file)
--- 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 1d251133379b8c03f2458e42c1c4ef044857334a..6d18db16c7c94b231aa23a15f3615ea70a1f2aed 100644 (file)
--- 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 d0cbe871ff87ddc8ca27ebf237906151412016b4..f3d05cac279d6a1759e5918074e3b0575bd75119 100644 (file)
--- 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 9f0c2dab09ce7f9fe98d6e48c8abc5c29eabb4dc..ed6f59a548f10e72e577dbabd1cb6fe6b8d555e8 100644 (file)
--- 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()