- Connection has execution_options(), generative method

which accepts keywords that affect how the statement
is executed w.r.t. the DBAPI.   Currently supports
"stream_results", causes psycopg2 to use a server
side cursor for that statement.   Can also be set
upon select() and text() constructs directly as well
as ORM Query().

5 files changed, 29 insertions, 18 deletions

diff --git a/lib/sqlalchemy/dialects/mysql/oursql.py b/lib/sqlalchemy/dialects/mysql/oursql.py
index 70bec53a3..dc1310db7 100644
--- a/lib/sqlalchemy/dialects/mysql/oursql.py
+++ b/lib/sqlalchemy/dialects/mysql/oursql.py
@@ -54,7 +54,7 @@ class MySQL_oursqlExecutionContext(MySQLExecutionContext):
     
     @property
     def plain_query(self):
-        return self._connection.options.get('plain_query', False)
+        return self.execution_options.get('_oursql_plain_query', False)
 
 
 class MySQL_oursql(MySQLDialect):
@@ -90,7 +90,7 @@ class MySQL_oursql(MySQLDialect):
         connection.cursor().execute('BEGIN', plain_query=True)
 
     def _xa_query(self, connection, query, xid):
-        connection._with_options(plain_query=True).execute(query % connection.connection._escape_string(xid))
+        connection.execution_options(_oursql_plain_query=True).execute(query % connection.connection._escape_string(xid))
 
     # Because mysql is bad, these methods have to be reimplemented to use _PlainQuery. Basically, some queries
     # refuse to return any data if they're run through the parameterized query API, or refuse to be parameterized
@@ -115,12 +115,12 @@ class MySQL_oursql(MySQLDialect):
         self._xa_query(connection, 'XA COMMIT "%s"', xid)
 
     def has_table(self, connection, table_name, schema=None):
-        return MySQLDialect.has_table(self, connection._with_options(plain_query=True), table_name, schema)
+        return MySQLDialect.has_table(self, connection.execution_options(_oursql_plain_query=True), table_name, schema)
 
     def _show_create_table(self, connection, table, charset=None,
                            full_name=None):
         return MySQLDialect._show_create_table(self, 
-            connection.contextual_connect(close_with_result=True)._with_options(plain_query=True), 
+            connection.contextual_connect(close_with_result=True).execution_options(_oursql_plain_query=True), 
             table, charset, full_name)
 
     def is_disconnect(self, e):
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index e74e00d84..3d192a9be 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -719,10 +719,10 @@ class Connection(Connectable):
     .. index::
       single: thread safety; Connection
     """
-    options = {}
+    _execution_options = util.frozendict()
     
     def __init__(self, engine, connection=None, close_with_result=False,
-                 _branch=False, _options=None):
+                 _branch=False, _execution_options=None):
         """Construct a new Connection.
 
         Connection objects are typically constructed by an
@@ -736,8 +736,8 @@ class Connection(Connectable):
         self.__savepoint_seq = 0
         self.__branch = _branch
         self.__invalid = False
-        if _options:
-            self.options = _options
+        if _execution_options:
+            self._execution_options = self._execution_options.union(_execution_options)
 
     def _branch(self):
         """Return a new Connection which references this Connection's
@@ -750,7 +750,7 @@ class Connection(Connectable):
 
         return self.engine.Connection(self.engine, self.__connection, _branch=True)
     
-    def _with_options(self, **opt):
+    def execution_options(self, **opt):
         """Add keyword options to a Connection generatively.
         
         Experimental.  May change the name/signature at 
@@ -763,7 +763,7 @@ class Connection(Connectable):
         """
         return self.engine.Connection(
                     self.engine, self.__connection,
-                     _branch=self.__branch, _options=opt)
+                     _branch=self.__branch, _execution_options=opt)
     
     @property
     def dialect(self):
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index bb3688597..6db655832 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -224,13 +224,13 @@ class DefaultDialect(base.Dialect):
 
 
 class DefaultExecutionContext(base.ExecutionContext):
-    execution_options = util.frozendict()
     
     def __init__(self, dialect, connection, compiled_sql=None, compiled_ddl=None, statement=None, parameters=None):
         self.dialect = dialect
         self._connection = self.root_connection = connection
         self.engine = connection.engine
-
+        self.execution_options = connection._execution_options
+        
         if compiled_ddl is not None:
             self.compiled = compiled = compiled_ddl
             if not dialect.supports_unicode_statements:
@@ -268,7 +268,8 @@ class DefaultExecutionContext(base.ExecutionContext):
             self.isinsert = compiled.isinsert
             self.isupdate = compiled.isupdate
             self.isdelete = compiled.isdelete
-            self.execution_options = compiled.statement._execution_options
+            self.execution_options =\
+                    compiled.statement._execution_options.union(self.execution_options)
 
             if not parameters:
                 self.compiled_parameters = [compiled.construct_params()]
diff --git a/lib/sqlalchemy/sql/expression.py b/lib/sqlalchemy/sql/expression.py
index f2ad4351a..eb64fd571 100644
--- a/lib/sqlalchemy/sql/expression.py
+++ b/lib/sqlalchemy/sql/expression.py
@@ -2194,16 +2194,13 @@ class _Executable(object):
     _execution_options = util.frozendict()
 
     @_generative
-    def execution_options(self, **kwargs):
+    def execution_options(self, **kw):
         """ Set non-SQL options for the statement, such as dialect-specific options.
 
         The options available are covered in the respective dialect's section.
 
         """
-        _execution_options = self._execution_options.copy()
-        for key, value in kwargs.items():
-            _execution_options[key] = value
-        self._execution_options = _execution_options
+        self._execution_options = self._execution_options.union(kw)
 
     
 class _TextClause(_Executable, ClauseElement):
@@ -2252,6 +2249,11 @@ class _TextClause(_Executable, ClauseElement):
         else:
             return None
 
+    def _generate(self):
+        s = self.__class__.__new__(self.__class__)
+        s.__dict__ = self.__dict__.copy()
+        return s
+
     def _copy_internals(self, clone=_clone):
         self.bindparams = dict((b.key, clone(b))
                                for b in self.bindparams.values())
diff --git a/lib/sqlalchemy/util.py b/lib/sqlalchemy/util.py
index cfa891554..c2ef81425 100644
--- a/lib/sqlalchemy/util.py
+++ b/lib/sqlalchemy/util.py
@@ -154,6 +154,14 @@ class frozendict(dict):
     def __init__(self, *args):
         pass
 
+    def union(self, d):
+        if not self:
+            return frozendict(d)
+        else:
+            d2 = self.copy()
+            d2.update(d)
+            return frozendict(d2)
+            
     def __repr__(self):
         return "frozendict(%s)" % dict.__repr__(self)