"load=False".
- many-to-one relations now fire off a lazyload in fewer cases, including
in most cases will not fetch the "old" value when a new one is replaced.
-
+ - added "make_transient()" helper function which transforms a persistent/
+ detached instance into a transient one (i.e. deletes the instance_key
+ and removes from any session.) [ticket:1052]
+
- sql
- returning() support is native to insert(), update(), delete(). Implementations
of varying levels of functionality exist for Postgresql, Firebird, MSSQL and
self.use_ansi = use_ansi
self.optimize_limits = optimize_limits
-# TODO: implement server_version_info for oracle
-# def initialize(self, connection):
-# super(OracleDialect, self).initialize(connection)
-# self.implicit_returning = self.server_version_info > (10, ) and \
-# self.__dict__.get('implicit_returning', True)
+ def initialize(self, connection):
+ super(OracleDialect, self).initialize(connection)
+ self.implicit_returning = self.server_version_info > (10, ) and \
+ self.__dict__.get('implicit_returning', True)
def do_release_savepoint(self, connection, name):
# Oracle does not support RELEASE SAVEPOINT
from sqlalchemy.orm.query import AliasOption, Query
from sqlalchemy.sql import util as sql_util
from sqlalchemy.orm.session import Session as _Session
-from sqlalchemy.orm.session import object_session, sessionmaker
+from sqlalchemy.orm.session import object_session, sessionmaker, make_transient
from sqlalchemy.orm.scoping import ScopedSession
from sqlalchemy import util as sa_util
'join',
'lazyload',
'mapper',
+ 'make_transient',
'noload',
'object_mapper',
'object_session',
return state
+def make_transient(instance):
+ """Make the given instance 'transient'.
+
+ This will remove its association with any
+ session and additionally will remove its "identity key",
+ such that it's as though the object were newly constructed,
+ except retaining its values.
+
+ """
+ state = attributes.instance_state(instance)
+ s = _state_session(state)
+ if s:
+ s._expunge_state(state)
+ del state.key
+
+
def object_session(instance):
"""Return the ``Session`` to which instance belongs, or None."""
return text
+ def _create_crud_bind_param(self, col, value):
+ bindparam = sql.bindparam(col.key, value, type_=col.type)
+ self.binds[col.key] = bindparam
+ return self.bindparam_string(self._truncate_bindparam(bindparam))
+
def _get_colparams(self, stmt):
"""create a set of tuples representing column/string pairs for use
in an INSERT or UPDATE statement.
"""
- def create_bind_param(col, value):
- bindparam = sql.bindparam(col.key, value, type_=col.type)
- self.binds[col.key] = bindparam
- return self.bindparam_string(self._truncate_bindparam(bindparam))
-
self.postfetch = []
self.prefetch = []
self.returning = []
# no parameters in the statement, no parameters in the
# compiled params - return binds for all columns
if self.column_keys is None and stmt.parameters is None:
- return [(c, create_bind_param(c, None)) for c in stmt.table.columns]
+ return [(c, self._create_crud_bind_param(c, None)) for c in stmt.table.columns]
# if we have statement parameters - set defaults in the
# compiled params
if c.key in parameters:
value = parameters[c.key]
if sql._is_literal(value):
- value = create_bind_param(c, value)
+ value = self._create_crud_bind_param(c, value)
else:
self.postfetch.append(c)
value = self.process(value.self_group())
values.append((c, self.process(c.default.arg.self_group())))
self.returning.append(c)
elif c.default is not None:
- values.append((c, create_bind_param(c, None)))
+ values.append((c, self._create_crud_bind_param(c, None)))
self.prefetch.append(c)
else:
self.returning.append(c)
) or \
self.dialect.preexecute_autoincrement_sequences:
- values.append((c, create_bind_param(c, None)))
+ values.append((c, self._create_crud_bind_param(c, None)))
self.prefetch.append(c)
elif isinstance(c.default, schema.ColumnDefault):
# dont add primary key column to postfetch
self.postfetch.append(c)
else:
- values.append((c, create_bind_param(c, None)))
+ values.append((c, self._create_crud_bind_param(c, None)))
self.prefetch.append(c)
elif c.server_default is not None:
if not c.primary_key:
values.append((c, self.process(c.onupdate.arg.self_group())))
self.postfetch.append(c)
else:
- values.append((c, create_bind_param(c, None)))
+ values.append((c, self._create_crud_bind_param(c, None)))
self.prefetch.append(c)
elif c.server_onupdate is not None:
self.postfetch.append(c)
from sqlalchemy.test.util import gc_collect
import inspect
import pickle
-from sqlalchemy.orm import create_session, sessionmaker, attributes
+from sqlalchemy.orm import create_session, sessionmaker, attributes, make_transient
import sqlalchemy as sa
from sqlalchemy.test import engines, testing, config
from sqlalchemy import Integer, String, Sequence
eq_(bind.connect().execute("select count(1) from users").scalar(), 1)
sess.close()
+ @testing.resolve_artifact_names
+ def test_make_transient(self):
+ mapper(User, users)
+ sess = create_session()
+ sess.add(User(name='test'))
+ sess.flush()
+
+ u1 = sess.query(User).first()
+ make_transient(u1)
+ assert u1 not in sess
+ sess.add(u1)
+ assert u1 in sess.new
+
+ u1 = sess.query(User).first()
+ sess.expunge(u1)
+ make_transient(u1)
+ sess.add(u1)
+ assert u1 in sess.new
+
@testing.resolve_artifact_names
def test_autoflush_expressions(self):
"""test that an expression which is dependent on object state is