- [feature] Added SQLite execution option

    "sqlite_raw_colnames=True", will bypass
    attempts to remove "." from column names
    returned by SQLite cursor.description.
    [ticket:2475]

5 files changed, 92 insertions, 26 deletions

diff --git a/CHANGES b/CHANGES
index 6b7d50c6f..1a2eb37f9 100644
--- a/CHANGES
+++ b/CHANGES
@@ -247,6 +247,12 @@ CHANGES
     Nathan Wright for the work and tests on
     this.  [ticket:2363]
 
+  - [feature] Added SQLite execution option
+    "sqlite_raw_colnames=True", will bypass
+    attempts to remove "." from column names
+    returned by SQLite cursor.description.
+    [ticket:2475] Also in 0.7.7.
+
 - mssql
   - [bug] removed legacy behavior whereby 
     a column comparison to a scalar SELECT via 
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index 754bf7966..39da0cc42 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -483,6 +483,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
@@ -493,6 +509,7 @@ class SQLiteDialect(default.DefaultDialect):
     supports_cast = True
 
     default_paramstyle = 'qmark'
+    execution_ctx_cls = SQLiteExecutionContext
     statement_compiler = SQLiteCompiler
     ddl_compiler = SQLiteDDLCompiler
     type_compiler = SQLiteTypeCompiler
@@ -567,16 +584,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 a2695e337..a631f5c32 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -2726,7 +2726,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
         self.case_sensitive = dialect.case_sensitive
 
         # high precedence key values.
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 1f72d005d..b0d2ab946 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,
@@ -361,6 +357,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 29a6ed355..31b408796 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()