from sqlalchemy import exc, schema, util, types, log
from sqlalchemy.sql import expression
-
class Dialect(object):
"""Define the behavior of a specific database and DB-API combination.
def _create_key_cache(self):
# local copies to avoid circular ref against 'self'
props = self.__props
- context = self.context
def lookup_key(key):
"""Given a key, which could be a ColumnElement, string, etc.,
matches it to the appropriate key we got from the result set's
import weakref
import sqlalchemy.exceptions as sa_exc
-import sqlalchemy.orm.attributes
from sqlalchemy import util, sql, engine
from sqlalchemy.sql import util as sql_util, expression
from sqlalchemy.orm import (
"""Return a new ``Query`` object corresponding to this ``Session``."""
return self._query_cls(entities, self, **kwargs)
+
def _autoflush(self):
if self.autoflush and (self.transaction is None or self.transaction.autoflush):
self.flush()
def expire_all(self):
"""Expires all persistent instances within this Session."""
+
for state in self.identity_map.all_states():
_expire_state(state, None)
from sqlalchemy.ext import declarative as decl
from testlib import sa, testing
-from testlib.sa import MetaData, Table, Column, Integer, String, ForeignKey, ForeignKeyConstraint
-from testlib.sa.orm import relation, create_session, class_mapper
+from testlib.sa import MetaData, Table, Column, Integer, String, ForeignKey, ForeignKeyConstraint, asc
+from testlib.sa.orm import relation, create_session, class_mapper, eagerload
from testlib.testing import eq_
from orm._base import ComparableEntity
a1 = sess.query(Address).filter(Address.email == 'two').one()
eq_(a1, Address(email='two'))
eq_(a1.user, User(name='u1'))
+
+ def test_eager_order_by(self):
+ class Address(Base, ComparableEntity):
+ __tablename__ = 'addresses'
+
+ id = Column('id', Integer, primary_key=True)
+ email = Column('email', String(50))
+ user_id = Column('user_id', Integer, ForeignKey('users.id'))
+ class User(Base, ComparableEntity):
+ __tablename__ = 'users'
+
+ id = Column('id', Integer, primary_key=True)
+ name = Column('name', String(50))
+ addresses = relation("Address", order_by=Address.email)
+
+ Base.metadata.create_all()
+ u1 = User(name='u1', addresses=[
+ Address(email='two'),
+ Address(email='one'),
+ ])
+ sess = create_session()
+ sess.save(u1)
+ sess.flush()
+ sess.clear()
+ eq_(sess.query(User).options(eagerload(User.addresses)).all(), [User(name='u1', addresses=[
+ Address(email='one'),
+ Address(email='two'),
+ ])])
+
+
def test_as_declarative(self):
class User(ComparableEntity):
__tablename__ = 'users'