needless updates of the primary key column during a so-called
"row switch" operation, i.e. add + delete of two objects
with the same PK.
+
+ - Now uses sqlalchemy.orm.exc.DetachedInstanceError when an
+ attribute load or refresh action fails due to object
+ being detached from any Session. UnboundExecutionError
+ is specific to engines bound to sessions and statements.
- sql
- Added math negation operator support, -x.
"""
from sqlalchemy import log, util
-import sqlalchemy.exceptions as sa_exc
+from sqlalchemy import exc as sa_exc
+from sqlalchemy.orm import exc as sa_exc
from sqlalchemy.sql import operators
from sqlalchemy.orm import (
attributes, object_session, util as mapperutil, strategies, object_mapper
if sess is None:
sess = object_session(instance)
if sess is None:
- raise sa_exc.UnboundExecutionError(
+ raise orm_exc.DetachedInstanceError(
"Parent instance %s is not bound to a Session, and no "
"contextual session is established; lazy load operation "
"of attribute '%s' cannot proceed" % (
class UnmappedError(sa.exc.InvalidRequestError):
"""TODO"""
-
+class DetachedInstanceError(sa.exc.SQLAlchemyError):
+ """An attempt to access unloaded attributes on a mapped instance that is detached."""
+
class UnmappedInstanceError(UnmappedError):
"""An mapping operation was requested for an unknown instance."""
from sqlalchemy import sql, util, log, exc as sa_exc
from sqlalchemy.sql import expression, visitors, operators, util as sqlutil
-from sqlalchemy.orm import attributes, exc, sync
+from sqlalchemy.orm import attributes, sync, exc as orm_exc
from sqlalchemy.orm.interfaces import (
MapperProperty, EXT_CONTINUE, PropComparator
)
except KeyError:
prop = self._props.get(column.key, None)
if prop:
- raise exc.UnmappedColumnError("Column '%s.%s' is not available, due to conflicting property '%s':%s" % (column.table.name, column.name, column.key, repr(prop)))
+ raise orm_exc.UnmappedColumnError("Column '%s.%s' is not available, due to conflicting property '%s':%s" % (column.table.name, column.name, column.key, repr(prop)))
else:
- raise exc.UnmappedColumnError("No column %s is configured on mapper %s..." % (column, self))
+ raise orm_exc.UnmappedColumnError("No column %s is configured on mapper %s..." % (column, self))
# TODO: improve names?
def _get_state_attr_by_column(self, state, column):
instance = uowtransaction.session.identity_map[instance_key]
existing = attributes.instance_state(instance)
if not uowtransaction.is_deleted(existing):
- raise exc.FlushError(
+ raise orm_exc.FlushError(
"New instance %s with identity key %s conflicts with persistent instance %s" %
(state_str(state), instance_key, state_str(existing)))
if self._should_log_debug:
if connection.dialect.supports_sane_rowcount:
if rows != len(update):
- raise exc.ConcurrentModificationError(
+ raise orm_exc.ConcurrentModificationError(
"Updated rowcount %d does not match number of objects updated %d" %
(rows, len(update)))
statement = table.delete(clause)
c = connection.execute(statement, del_objects)
if c.supports_sane_multi_rowcount() and c.rowcount != len(del_objects):
- raise exc.ConcurrentModificationError("Deleted rowcount %d does not match "
+ raise orm_exc.ConcurrentModificationError("Deleted rowcount %d does not match "
"number of objects deleted %d" % (c.rowcount, len(del_objects)))
for state, mapper, connection in tups:
if not currentload and version_id_col is not None and context.version_check and \
self._get_state_attr_by_column(state, self.version_id_col) != row[version_id_col]:
- raise exc.ConcurrentModificationError(
+ raise orm_exc.ConcurrentModificationError(
"Instance '%s' version of %s does not match %s"
% (state_str(state), self._get_state_attr_by_column(state, self.version_id_col), row[version_id_col]))
elif refresh_state:
mapper = _state_mapper(state)
session = _state_session(state)
if not session:
- raise sa_exc.UnboundExecutionError("Instance %s is not bound to a Session; "
+ raise orm_exc.DetachedInstanceError("Instance %s is not bound to a Session; "
"attribute refresh operation cannot proceed" % (state_str(state)))
has_key = _state_has_identity(state)
# if instance is pending, a refresh operation may not complete (even if PK attributes are assigned)
if has_key and result is None:
- raise exc.ObjectDeletedError("Instance '%s' has been deleted." % state_str(state))
+ raise orm_exc.ObjectDeletedError("Instance '%s' has been deleted." % state_str(state))
"""sqlalchemy.orm.interfaces.LoaderStrategy implementations, and related MapperOptions."""
-import sqlalchemy.exceptions as sa_exc
+from sqlalchemy import exc as sa_exc
from sqlalchemy import sql, util, log
from sqlalchemy.sql import util as sql_util
from sqlalchemy.sql import visitors, expression, operators
-from sqlalchemy.orm import mapper, attributes, interfaces
+from sqlalchemy.orm import mapper, attributes, interfaces, exc as orm_exc
from sqlalchemy.orm.interfaces import (
LoaderStrategy, StrategizedOption, MapperOption, PropertyOption,
serialize_path, deserialize_path, StrategizedProperty
session = sessionlib._state_session(state)
if session is None:
- raise sa_exc.UnboundExecutionError(
+ raise orm_exc.DetachedInstanceError(
"Parent instance %s is not bound to a Session; "
"deferred load operation of attribute '%s' cannot proceed" %
(mapperutil.state_str(state), self.key)
session = sessionlib._state_session(state)
if session is None:
- raise sa_exc.UnboundExecutionError(
+ raise orm_exc.DetachedInstanceError(
"Parent instance %s is not bound to a Session; "
"lazy load operation of attribute '%s' cannot proceed" %
(mapperutil.state_str(state), self.key)
from sqlalchemy import Integer, String, ForeignKey, exc as sa_exc
from sqlalchemy.test.schema import Table
from sqlalchemy.test.schema import Column
-from sqlalchemy.orm import mapper, relation, create_session, attributes, deferred
+from sqlalchemy.orm import mapper, relation, create_session, attributes, deferred, exc as orm_exc
from test.orm import _base, _fixtures
u = s.query(User).get(7)
s.expunge_all()
- assert_raises_message(sa.exc.InvalidRequestError, r"is not persistent within this Session", s.expire, u)
+ assert_raises_message(sa_exc.InvalidRequestError, r"is not persistent within this Session", s.expire, u)
@testing.resolve_artifact_names
def test_get_refreshes(self):
sess.expire(u, attribute_names=['name'])
sess.expunge(u)
- assert_raises(sa.exc.UnboundExecutionError, getattr, u, 'name')
+ assert_raises(orm_exc.DetachedInstanceError, getattr, u, 'name')
@testing.resolve_artifact_names
def test_pending_raises(self):
sess = create_session()
u = User(id=15)
sess.add(u)
- assert_raises(sa.exc.InvalidRequestError, sess.expire, u, ['name'])
+ assert_raises(sa_exc.InvalidRequestError, sess.expire, u, ['name'])
@testing.resolve_artifact_names
def test_no_instance_key(self):
s = create_session()
u = s.query(User).get(7)
s.expunge_all()
- assert_raises_message(sa.exc.InvalidRequestError, r"is not persistent within this Session", lambda: s.refresh(u))
+ assert_raises_message(sa_exc.InvalidRequestError, r"is not persistent within this Session", lambda: s.refresh(u))
@testing.resolve_artifact_names
def test_refresh_expired(self):
from sqlalchemy.test.testing import assert_raises, assert_raises_message
import datetime
from sqlalchemy import exc as sa_exc
-from sqlalchemy.orm import attributes
+from sqlalchemy.orm import attributes, exc as orm_exc
import sqlalchemy as sa
from sqlalchemy.test import testing
from sqlalchemy import Integer, String, ForeignKey, SmallInteger
q = sess.query(User)
u = q.filter(users.c.id == 7).first()
sess.expunge(u)
- assert_raises(sa_exc.InvalidRequestError, getattr, u, 'addresses')
+ assert_raises(orm_exc.DetachedInstanceError, getattr, u, 'addresses')
@testing.resolve_artifact_names
def test_orderby(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
