as appropriate for more complex situations.
[ticket:1628]
- - Added "execution_options()" to Selects, which set statement
- specific options. These enable e.g. dialect specific options
- such as whether to enable using server side cursors, etc.
+ - Added "execution_options()" to select()/text(), which set the
+ default options for the Connection. See the note in "engines".
- Deprecated or removed:
* "scalar" flag on select() is removed, use
create_engine(... isolation_level="..."); available on
postgresql and sqlite. [ticket:443]
+ - 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().
+
- fixed the import for entrypoint-driven dialects to
not rely upon silly tb_info trick to determine import
error status. [ticket:1630]
@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):
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
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):
.. 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
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
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
"""
return self.engine.Connection(
self.engine, self.__connection,
- _branch=self.__branch, _options=opt)
+ _branch=self.__branch, _execution_options=opt)
@property
def dialect(self):
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:
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()]
_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):
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())
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)
# ... but enabled for this one.
assert result.cursor.name
+ # and this one
+ result = engine.connect().execution_options(stream_results=True).execute("select 1")
+ assert result.cursor.name
+
+ # not this one
+ result = engine.connect().execution_options(stream_results=False).execute(s)
+ assert not result.cursor.name
+
def test_ss_explicitly_disabled(self):
s = select([1]).execution_options(stream_results=False)
result = ss_engine.execute(s)
s = text('select 42')
result = engine.execute(s)
assert not result.cursor.name
- s = text('select 42', execution_options=dict(stream_results=True))
+ s = text('select 42').execution_options(stream_results=True)
result = engine.execute(s)
assert result.cursor.name
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
