SQLAlchemy object-relational configuration involves the usage of Table,
mapper(), and class objects to define the three areas of configuration.
-declarative moves these three types of configuration underneath the
-individual mapped class. Regular SQLAlchemy schema and ORM constructs are
-used in most cases::
+declarative moves these three types of configuration underneath the individual
+mapped class. Regular SQLAlchemy schema and ORM constructs are used in most
+cases::
from sqlalchemy.ext.declarative import declarative_base
id = Column('id', Integer, primary_key=True)
name = Column('name', String(50))
-Above, the ``declarative_base`` callable produces a new base class from
-which all mapped classes inherit from. When the class definition is
-completed, a new ``Table`` and ``mapper()`` have been generated, accessible
-via the ``__table__`` and ``__mapper__`` attributes on the ``SomeClass``
-class.
+Above, the ``declarative_base`` callable produces a new base class from which
+all mapped classes inherit from. When the class definition is completed, a
+new ``Table`` and ``mapper()`` have been generated, accessible via the
+``__table__`` and ``__mapper__`` attributes on the ``SomeClass`` class.
You may omit the names from the Column definitions. Declarative will fill
them in for you::
id = Column(Integer, primary_key=True)
name = Column(String(50))
-Attributes may be added to the class after its construction, and they will
-be added to the underlying ``Table`` and ``mapper()`` definitions as
+Attributes may be added to the class after its construction, and they will be
+added to the underlying ``Table`` and ``mapper()`` definitions as
appropriate::
SomeClass.data = Column('data', Unicode)
SomeClass.related = relation(RelatedInfo)
Classes which are mapped explicitly using ``mapper()`` can interact freely
-with declarative classes.
+with declarative classes.
-The ``declarative_base`` base class contains a
-``MetaData`` object where newly defined ``Table`` objects are collected.
-This is accessed via the ``metadata`` class level accessor, so to
-create tables we can say::
+The ``declarative_base`` base class contains a ``MetaData`` object where newly
+defined ``Table`` objects are collected. This is accessed via the
+``metadata`` class level accessor, so to create tables we can say::
engine = create_engine('sqlite://')
Base.metadata.create_all(engine)
-The ``Engine`` created above may also be directly associated with the
-declarative base class using the ``engine`` keyword argument, where it will
-be associated with the underlying ``MetaData`` object and allow SQL
-operations involving that metadata and its tables to make use of that
-engine automatically::
+The ``Engine`` created above may also be directly associated with the
+declarative base class using the ``engine`` keyword argument, where it will be
+associated with the underlying ``MetaData`` object and allow SQL operations
+involving that metadata and its tables to make use of that engine
+automatically::
Base = declarative_base(engine=create_engine('sqlite://'))
Base.metadata.bind = create_engine('sqlite://')
-The ``declarative_base`` can also receive a pre-created ``MetaData``
-object, which allows a declarative setup to be associated with an already existing traditional collection of ``Table`` objects::
+The ``declarative_base`` can also receive a pre-created ``MetaData`` object,
+which allows a declarative setup to be associated with an already existing
+traditional collection of ``Table`` objects::
mymetadata = MetaData()
Base = declarative_base(metadata=mymetadata)
Relations to other classes are done in the usual way, with the added feature
that the class specified to ``relation()`` may be a string name. The "class
registry" associated with ``Base`` is used at mapper compilation time to
-resolve the name into the actual class object, which is expected to have
-been defined once the mapper configuration is used::
+resolve the name into the actual class object, which is expected to have been
+defined once the mapper configuration is used::
class User(Base):
__tablename__ = 'users'
email = Column(String(50))
user_id = Column(Integer, ForeignKey('users.id'))
-Column constructs, since they are just that, are immediately usable, as
-below where we define a primary join condition on the ``Address`` class
-using them::
+Column constructs, since they are just that, are immediately usable, as below
+where we define a primary join condition on the ``Address`` class using them::
class Address(Base)
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True)
email = Column(String(50))
user_id = Column(Integer, ForeignKey('users.id'))
- user = relation(User, primaryjoin=user_id==User.id)
-
-When an explicit join condition or other configuration which depends
-on multiple classes cannot be defined immediately due to some classes
-not yet being available, these can be defined after all classes have
-been created. Attributes which are added to the class after
-its creation are associated with the Table/mapping in the same
-way as if they had been defined inline::
-
- User.addresses = relation(Address, primaryjoin=Address.user_id==User.id)
-
-Synonyms are one area where ``declarative`` needs to slightly change the
-usual SQLAlchemy configurational syntax. To define a getter/setter which
-proxies to an underlying attribute, use ``synonym`` with the ``descriptor``
-argument::
+ user = relation(User, primaryjoin=user_id == User.id)
+
+When an explicit join condition or other configuration which depends on
+multiple classes cannot be defined immediately due to some classes not yet
+being available, these can be defined after all classes have been created.
+Attributes which are added to the class after its creation are associated with
+the Table/mapping in the same way as if they had been defined inline::
+
+ User.addresses = relation(Address, primaryjoin=Address.user_id == User.id)
+
+Synonyms are one area where ``declarative`` needs to slightly change the usual
+SQLAlchemy configurational syntax. To define a getter/setter which proxies to
+an underlying attribute, use ``synonym`` with the ``descriptor`` argument::
class MyClass(Base):
__tablename__ = 'sometable'
session.query(MyClass).filter(MyClass.attr == 'some other value').all()
As an alternative to ``__tablename__``, a direct ``Table`` construct may be
-used. The ``Column`` objects, which in this case require their names,
-will be added to the mapping just like a regular mapping to a table::
+used. The ``Column`` objects, which in this case require their names, will be
+added to the mapping just like a regular mapping to a table::
class MyClass(Base):
__table__ = Table('my_table', Base.metadata,
__table__ = Table('my_table', Base.metadata, autoload=True)
Mapper arguments are specified using the ``__mapper_args__`` class variable.
-Note that the column objects declared on the class are immediately usable,
-as in this joined-table inheritance example::
+Note that the column objects declared on the class are immediately usable, as
+in this joined-table inheritance example::
class Person(Base):
__tablename__ = 'people'
id = Column(Integer, primary_key=True)
discriminator = Column(String(50))
- __mapper_args__ = {'polymorphic_on':discriminator}
+ __mapper_args__ = {'polymorphic_on': discriminator}
class Engineer(Person):
__tablename__ = 'engineers'
- __mapper_args__ = {'polymorphic_identity':'engineer'}
+ __mapper_args__ = {'polymorphic_identity': 'engineer'}
id = Column(Integer, ForeignKey('people.id'), primary_key=True)
primary_language = Column(String(50))
For single-table inheritance, the ``__tablename__`` and ``__table__`` class
-variables are optional on a class when the class inherits from another
-mapped class.
+variables are optional on a class when the class inherits from another mapped
+class.
As a convenience feature, the ``declarative_base()`` sets a default
-constructor on classes which takes keyword arguments, and assigns them to
-the named attributes::
+constructor on classes which takes keyword arguments, and assigns them to the
+named attributes::
e = Engineer(primary_language='python')
Note that ``declarative`` has no integration built in with sessions, and is
-only intended as an optional syntax for the regular usage of mappers and
-Table objects. A typical application setup using ``scoped_session`` might
-look like::
+only intended as an optional syntax for the regular usage of mappers and Table
+objects. A typical application setup using ``scoped_session`` might look
+like::
engine = create_engine('postgres://scott:tiger@localhost/test')
- Session = scoped_session(sessionmaker(transactional=True, autoflush=False, bind=engine))
+ Session = scoped_session(sessionmaker(transactional=True,
+ autoflush=False,
+ bind=engine))
Base = declarative_base()
Mapped instances then make usage of ``Session`` in the usual way.
-"""
+"""
from sqlalchemy.schema import Table, Column, MetaData
from sqlalchemy.orm import synonym as _orm_synonym, mapper, comparable_property
from sqlalchemy.orm.interfaces import MapperProperty
from sqlalchemy.sql import util as sql_util
-__all__ = ['declarative_base', 'synonym_for', 'comparable_using',
- 'declared_synonym']
+__all__ = 'declarative_base', 'synonym_for', 'comparable_using'
class DeclarativeMeta(type):
continue
prop = _deferred_relation(cls, value)
our_stuff[k] = prop
-
+
# set up attributes in the order they were created
- our_stuff.sort(lambda x, y: cmp(our_stuff[x]._creation_order, our_stuff[y]._creation_order))
+ our_stuff.sort(lambda x, y: cmp(our_stuff[x]._creation_order,
+ our_stuff[y]._creation_order))
table = None
if '__table__' not in cls.__dict__:
if inherits:
mapper_args['inherits'] = inherits
if not mapper_args.get('concrete', False) and table:
- # figure out the inherit condition with relaxed rules about nonexistent tables,
- # to allow for ForeignKeys to not-yet-defined tables (since we know for sure that our parent
- # table is defined within the same MetaData)
- mapper_args['inherit_condition'] = sql_util.join_condition(inherits.__table__, table, ignore_nonexistent_tables=True)
-
+ # figure out the inherit condition with relaxed rules
+ # about nonexistent tables, to allow for ForeignKeys to
+ # not-yet-defined tables (since we know for sure that our
+ # parent table is defined within the same MetaData)
+ mapper_args['inherit_condition'] = sql_util.join_condition(
+ inherits.__table__, table,
+ ignore_nonexistent_tables=True)
+
if hasattr(cls, '__mapper_cls__'):
mapper_cls = util.unbound_method_to_callable(cls.__mapper_cls__)
else:
mapper_cls = mapper
- cls.__mapper__ = mapper_cls(cls, table, properties=our_stuff, **mapper_args)
+ cls.__mapper__ = mapper_cls(cls, table, properties=our_stuff,
+ **mapper_args)
return type.__init__(cls, classname, bases, dict_)
def __setattr__(cls, key, value):
type.__setattr__(cls, key, value)
def _deferred_relation(cls, prop):
- if isinstance(prop, PropertyLoader) and isinstance(prop.argument, basestring):
+ if (isinstance(prop, PropertyLoader) and
+ isinstance(prop.argument, basestring)):
arg = prop.argument
def return_cls():
try:
return cls._decl_class_registry[arg]
except KeyError:
- raise exceptions.InvalidRequestError("When compiling mapper %s, could not locate a declarative class named %r. Consider adding this property to the %r class after both dependent classes have been defined." % (prop.parent, arg, prop.parent.class_))
+ raise exceptions.InvalidRequestError(
+ "When compiling mapper %s, could not locate a declarative "
+ "class named %r. Consider adding this property to the %r "
+ "class after both dependent classes have been defined." % (
+ prop.parent, arg, prop.parent.class_))
prop.argument = return_cls
return prop
-def declared_synonym(prop, name):
- """Deprecated. Use synonym(name, descriptor=prop)."""
- return _orm_synonym(name, descriptor=prop)
-declared_synonym = util.deprecated(None, False)(declared_synonym)
-
def synonym_for(name, map_column=False):
"""Decorator, make a Python @property a query synonym for a column.
def prop(self):
return 'special sauce'
- The regular ``synonym()`` is also usable directly in a declarative
- setting and may be convenient for read/write properties::
+ The regular ``synonym()`` is also usable directly in a declarative setting
+ and may be convenient for read/write properties::
prop = synonym('col', descriptor=property(_read_prop, _write_prop))
declarative setting and may be convenient for read/write properties::
prop = comparable_property(MyComparatorType)
+
"""
def decorate(fn):
return comparable_property(comparator_factory, fn)
def __init__(self, **kwargs):
for k in kwargs:
if not hasattr(type(self), k):
- raise TypeError('%r is an invalid keyword argument for %s' %
- (k, type(self).__name__))
+ raise TypeError(
+ "%r is an invalid keyword argument for %s" %
+ (k, type(self).__name__))
setattr(self, k, kwargs[k])
return Base
import testenv; testenv.configure_for_tests()
-from sqlalchemy import *
-from sqlalchemy.orm import *
-from sqlalchemy.orm.interfaces import MapperExtension
-from sqlalchemy.ext.declarative import declarative_base, declared_synonym, \
- synonym_for, comparable_using
-from sqlalchemy import exc
-from testlib.fixtures import Base as Fixture
-from testlib import *
+from sqlalchemy.ext import declarative as decl
+from testlib import sa, testing
+from testlib.sa import MetaData, Table, Column, Integer, String, ForeignKey
+from testlib.sa.orm import relation, create_session
+from testlib.testing import eq_
+from orm._base import ComparableEntity
-class DeclarativeTest(TestBase, AssertsExecutionResults):
+class DeclarativeTest(testing.TestBase, testing.AssertsExecutionResults):
def setUp(self):
global Base
- Base = declarative_base(testing.db)
+ Base = decl.declarative_base(testing.db)
def tearDown(self):
Base.metadata.drop_all()
def test_basic(self):
- class User(Base, Fixture):
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True)
name = Column('name', String(50))
addresses = relation("Address", backref="user")
- class Address(Base, Fixture):
+ class Address(Base, ComparableEntity):
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True)
Base.metadata.create_all()
- assert Address.__table__.c['id'].name == 'id'
- assert Address.__table__.c['_email'].name == 'email'
- assert Address.__table__.c['_user_id'].name == 'user_id'
+ eq_(Address.__table__.c['id'].name, 'id')
+ eq_(Address.__table__.c['_email'].name, 'email')
+ eq_(Address.__table__.c['_user_id'].name, 'user_id')
u1 = User(name='u1', addresses=[
Address(email='one'),
sess.flush()
sess.clear()
- self.assertEquals(sess.query(User).all(), [User(name='u1', addresses=[
+ eq_(sess.query(User).all(), [User(name='u1', addresses=[
Address(email='one'),
Address(email='two'),
])])
- a1 = sess.query(Address).filter(Address.email=='two').one()
- self.assertEquals(a1, Address(email='two'))
- self.assertEquals(a1.user, User(name='u1'))
-
+ a1 = sess.query(Address).filter(Address.email == 'two').one()
+ eq_(a1, Address(email='two'))
+ eq_(a1.user, User(name='u1'))
+
def test_recompile_on_othermapper(self):
"""declarative version of the same test in mappers.py"""
id = Column('id', Integer, primary_key=True)
email = Column('email', String(50))
user_id = Column('user_id', Integer, ForeignKey('users.id'))
- user = relation("User", primaryjoin=user_id==User.id, backref="addresses")
+ user = relation("User", primaryjoin=user_id == User.id,
+ backref="addresses")
assert mapperlib._new_mappers is True
u = User()
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True)
addresses = relation("Address")
-
+
def go():
class Address(Base):
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True)
- foo = column_property(User.id==5)
- self.assertRaises(exc.InvalidRequestError, go)
-
+ foo = sa.orm.column_property(User.id == 5)
+ self.assertRaises(sa.exc.InvalidRequestError, go)
+
def test_add_prop(self):
- class User(Base, Fixture):
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True)
User.name = Column('name', String(50))
User.addresses = relation("Address", backref="user")
- class Address(Base, Fixture):
+ class Address(Base, ComparableEntity):
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True)
Base.metadata.create_all()
- assert Address.__table__.c['id'].name == 'id'
- assert Address.__table__.c['_email'].name == 'email'
- assert Address.__table__.c['_user_id'].name == 'user_id'
+ eq_(Address.__table__.c['id'].name, 'id')
+ eq_(Address.__table__.c['_email'].name, 'email')
+ eq_(Address.__table__.c['_user_id'].name, 'user_id')
u1 = User(name='u1', addresses=[
Address(email='one'),
sess.flush()
sess.clear()
- self.assertEquals(sess.query(User).all(), [User(name='u1', addresses=[
+ eq_(sess.query(User).all(), [User(name='u1', addresses=[
Address(email='one'),
Address(email='two'),
])])
- a1 = sess.query(Address).filter(Address.email=='two').one()
- self.assertEquals(a1, Address(email='two'))
- self.assertEquals(a1.user, User(name='u1'))
+ a1 = sess.query(Address).filter(Address.email == 'two').one()
+ eq_(a1, Address(email='two'))
+ eq_(a1.user, User(name='u1'))
+
-
def test_custom_mapper(self):
- class MyExt(MapperExtension):
+ class MyExt(sa.orm.MapperExtension):
def create_instance(self):
return "CHECK"
def mymapper(cls, tbl, **kwargs):
kwargs['extension'] = MyExt()
- return mapper(cls, tbl, **kwargs)
+ return sa.orm.mapper(cls, tbl, **kwargs)
from sqlalchemy.orm.mapper import Mapper
class MyMapper(Mapper):
ss_mapper = ss.mapper
for mapperfunc in (mymapper, MyMapper, ss_mapper):
- base = declarative_base()
+ base = decl.declarative_base()
class Foo(base):
__tablename__ = 'foo'
__mapper_cls__ = mapperfunc
id = Column(Integer, primary_key=True)
- assert Foo.__mapper__.compile().extension.create_instance() == 'CHECK'
+ eq_(Foo.__mapper__.compile().extension.create_instance(), 'CHECK')
- base = declarative_base(mapper=mapperfunc)
+ base = decl.declarative_base(mapper=mapperfunc)
class Foo(base):
__tablename__ = 'foo'
id = Column(Integer, primary_key=True)
- assert Foo.__mapper__.compile().extension.create_instance() == 'CHECK'
+ eq_(Foo.__mapper__.compile().extension.create_instance(), 'CHECK')
@testing.emits_warning('Ignoring declarative-like tuple value of '
'attribute id')
def test_oops(self):
def define():
- class User(Base, Fixture):
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True),
name = Column('name', String(50))
assert False
self.assertRaisesMessage(
- exc.ArgumentError,
+ sa.exc.ArgumentError,
"Mapper Mapper|User|users could not assemble any primary key",
define)
def test_expression(self):
- class User(Base, Fixture):
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True)
name = Column('name', String(50))
addresses = relation("Address", backref="user")
- class Address(Base, Fixture):
+ 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'))
- User.address_count = column_property(select([func.count(Address.id)]).where(Address.user_id==User.id).as_scalar())
+ User.address_count = sa.orm.column_property(
+ sa.select([sa.func.count(Address.id)]).
+ where(Address.user_id == User.id).as_scalar())
Base.metadata.create_all()
sess.flush()
sess.clear()
- self.assertEquals(sess.query(User).all(), [User(name='u1', address_count=2, addresses=[
- Address(email='one'),
- Address(email='two'),
- ])])
+ eq_(sess.query(User).all(),
+ [User(name='u1', address_count=2, addresses=[
+ Address(email='one'),
+ Address(email='two')])])
def test_column(self):
- class User(Base, Fixture):
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True)
Base.metadata.create_all()
u1 = User(name='u1', a='a', b='b')
- assert u1.a == 'a'
- assert User.a.get_history(u1) == (['a'], [], [])
+ eq_(u1.a, 'a')
+ eq_(User.a.get_history(u1), (['a'], [], []))
sess = create_session()
sess.save(u1)
sess.flush()
sess.clear()
- self.assertEquals(sess.query(User).all(),
- [User(name='u1', a='a', b='b')])
+ eq_(sess.query(User).all(),
+ [User(name='u1', a='a', b='b')])
def test_column_properties(self):
-
- class Address(Base, Fixture):
+ class Address(Base, ComparableEntity):
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True)
email = Column(String(50))
user_id = Column(Integer, ForeignKey('users.id'))
-
- class User(Base, Fixture):
+
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True)
name = Column('name', String(50))
- adr_count = column_property(select([func.count(Address.id)], Address.user_id==id).as_scalar())
+ adr_count = sa.orm.column_property(
+ sa.select([sa.func.count(Address.id)], Address.user_id == id).
+ as_scalar())
addresses = relation(Address)
-
+
Base.metadata.create_all()
-
+
u1 = User(name='u1', addresses=[
Address(email='one'),
Address(email='two'),
sess.flush()
sess.clear()
- self.assertEquals(sess.query(User).all(), [User(name='u1', adr_count=2, addresses=[
- Address(email='one'),
- Address(email='two'),
- ])])
+ eq_(sess.query(User).all(),
+ [User(name='u1', adr_count=2, addresses=[
+ Address(email='one'),
+ Address(email='two')])])
def test_column_properties_2(self):
-
- class Address(Base, Fixture):
+ class Address(Base, ComparableEntity):
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True)
email = Column(String(50))
user_id = Column(Integer, ForeignKey('users.id'))
- class User(Base, Fixture):
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True)
name = Column('name', String(50))
- # this is not "valid" but we want to test that Address.id doesnt get stuck into user's table
+ # this is not "valid" but we want to test that Address.id doesnt
+ # get stuck into user's table
adr_count = Address.id
-
- self.assertEquals(set(User.__table__.c.keys()), set(['id', 'name']))
- self.assertEquals(set(Address.__table__.c.keys()), set(['id', 'email', 'user_id']))
-
+
+ eq_(set(User.__table__.c.keys()), set(['id', 'name']))
+ eq_(set(Address.__table__.c.keys()), set(['id', 'email', 'user_id']))
+
def test_deferred(self):
- class User(Base, Fixture):
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
- name = deferred(Column(String(50)))
-
+ name = sa.orm.deferred(Column(String(50)))
+
Base.metadata.create_all()
sess = create_session()
sess.save(User(name='u1'))
sess.flush()
sess.clear()
-
- u1 = sess.query(User).filter(User.name=='u1').one()
+
+ u1 = sess.query(User).filter(User.name == 'u1').one()
assert 'name' not in u1.__dict__
def go():
- assert u1.name == 'u1'
+ eq_(u1.name, 'u1')
self.assert_sql_count(testing.db, go, 1)
-
- def test_synonym_inline(self):
- class User(Base, Fixture):
- __tablename__ = 'users'
-
- id = Column('id', Integer, primary_key=True)
- _name = Column('name', String(50))
- def _set_name(self, name):
- self._name = "SOMENAME " + name
- def _get_name(self):
- return self._name
- name = synonym('_name', descriptor=property(_get_name, _set_name))
-
- Base.metadata.create_all()
-
- sess = create_session()
- u1 = User(name='someuser')
- assert u1.name == "SOMENAME someuser", u1.name
- sess.save(u1)
- sess.flush()
- self.assertEquals(sess.query(User).filter(User.name=="SOMENAME someuser").one(), u1)
- @testing.uses_deprecated('Call to deprecated function declared_synonym')
- def test_decl_synonym_inline(self):
- class User(Base, Fixture):
+ def test_synonym_inline(self):
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True)
self._name = "SOMENAME " + name
def _get_name(self):
return self._name
- name = declared_synonym(property(_get_name, _set_name), '_name')
+ name = sa.orm.synonym('_name',
+ descriptor=property(_get_name, _set_name))
Base.metadata.create_all()
sess = create_session()
u1 = User(name='someuser')
- assert u1.name == "SOMENAME someuser", u1.name
+ eq_(u1.name, "SOMENAME someuser")
sess.save(u1)
sess.flush()
- self.assertEquals(sess.query(User).filter(User.name=="SOMENAME someuser").one(), u1)
+ eq_(sess.query(User).filter(User.name == "SOMENAME someuser").one(), u1)
def test_synonym_added(self):
- class User(Base, Fixture):
- __tablename__ = 'users'
-
- id = Column('id', Integer, primary_key=True)
- _name = Column('name', String(50))
- def _set_name(self, name):
- self._name = "SOMENAME " + name
- def _get_name(self):
- return self._name
- name = property(_get_name, _set_name)
- User.name = synonym('_name', descriptor=User.name)
-
- Base.metadata.create_all()
-
- sess = create_session()
- u1 = User(name='someuser')
- assert u1.name == "SOMENAME someuser", u1.name
- sess.save(u1)
- sess.flush()
- self.assertEquals(sess.query(User).filter(User.name=="SOMENAME someuser").one(), u1)
-
- @testing.uses_deprecated('Call to deprecated function declared_synonym')
- def test_decl_synonym_added(self):
- class User(Base, Fixture):
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True)
def _get_name(self):
return self._name
name = property(_get_name, _set_name)
- User.name = declared_synonym(User.name, '_name')
+ User.name = sa.orm.synonym('_name', descriptor=User.name)
Base.metadata.create_all()
sess = create_session()
u1 = User(name='someuser')
- assert u1.name == "SOMENAME someuser", u1.name
+ eq_(u1.name, "SOMENAME someuser")
sess.save(u1)
sess.flush()
- self.assertEquals(sess.query(User).filter(User.name=="SOMENAME someuser").one(), u1)
+ eq_(sess.query(User).filter(User.name == "SOMENAME someuser").one(), u1)
def test_joined_inheritance(self):
- class Company(Base, Fixture):
+ class Company(Base, ComparableEntity):
__tablename__ = 'companies'
id = Column('id', Integer, primary_key=True)
name = Column('name', String(50))
employees = relation("Person")
- class Person(Base, Fixture):
+ class Person(Base, ComparableEntity):
__tablename__ = 'people'
id = Column('id', Integer, primary_key=True)
- company_id = Column('company_id', Integer, ForeignKey('companies.id'))
+ company_id = Column('company_id', Integer,
+ ForeignKey('companies.id'))
name = Column('name', String(50))
discriminator = Column('type', String(50))
__mapper_args__ = {'polymorphic_on':discriminator}
sess.flush()
sess.clear()
- self.assertEquals(sess.query(Company).filter(Company.employees.of_type(Engineer).any(Engineer.primary_language=='cobol')).first(), c2)
-
+ eq_((sess.query(Company).
+ filter(Company.employees.of_type(Engineer).
+ any(Engineer.primary_language == 'cobol')).first()),
+ c2)
+
def test_inheritance_with_undefined_relation(self):
class Parent(Base):
__tablename__ = 'parent'
tp = Column('type', String(50))
__mapper_args__ = dict(polymorphic_on = tp)
-
+
class Child1(Parent):
__tablename__ = 'child1'
id = Column('id', Integer, ForeignKey('parent.id'), primary_key=True)
related_child2 = Column('c2', Integer, ForeignKey('child2.id'))
__mapper_args__ = dict(polymorphic_identity = 'child1')
-
- # no exception is raised by the ForeignKey to "child2" even though child2 doesn't exist yet
-
+
+ # no exception is raised by the ForeignKey to "child2" even though
+ # child2 doesn't exist yet
+
class Child2(Parent):
__tablename__ = 'child2'
id = Column('id', Integer, ForeignKey('parent.id'), primary_key=True)
related_child1 = Column('c1', Integer)
__mapper_args__ = dict(polymorphic_identity = 'child2')
-
- compile_mappers() # no exceptions here
-
+
+ sa.orm.compile_mappers() # no exceptions here
+
def test_reentrant_compile_via_foreignkey(self):
-
- class User(Base, Fixture):
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True)
name = Column('name', String(50))
addresses = relation("Address", backref="user")
- class Address(Base, Fixture):
+ 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(User.id))
-
- compile_mappers() # this forces a re-entrant compile() due to the User.id within the ForeignKey
+
+ # this forces a re-entrant compile() due to the User.id within the
+ # ForeignKey
+ sa.orm.compile_mappers()
Base.metadata.create_all()
u1 = User(name='u1', addresses=[
sess.flush()
sess.clear()
- self.assertEquals(sess.query(User).all(), [User(name='u1', addresses=[
+ eq_(sess.query(User).all(), [User(name='u1', addresses=[
Address(email='one'),
Address(email='two'),
])])
-
+
def test_relation_reference(self):
- class Address(Base, Fixture):
+ 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, Fixture):
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True)
name = Column('name', String(50))
addresses = relation("Address", backref="user",
- primaryjoin=id==Address.user_id)
+ primaryjoin=id == Address.user_id)
- User.address_count = column_property(select([func.count(Address.id)]).where(Address.user_id==User.id).as_scalar())
+ User.address_count = sa.orm.column_property(
+ sa.select([sa.func.count(Address.id)]).
+ where(Address.user_id == User.id).as_scalar())
Base.metadata.create_all()
sess.flush()
sess.clear()
- self.assertEquals(sess.query(User).all(), [User(name='u1', address_count=2, addresses=[
- Address(email='one'),
- Address(email='two'),
- ])])
+ eq_(sess.query(User).all(),
+ [User(name='u1', address_count=2, addresses=[
+ Address(email='one'),
+ Address(email='two')])])
def test_single_inheritance(self):
- class Company(Base, Fixture):
+ class Company(Base, ComparableEntity):
__tablename__ = 'companies'
id = Column('id', Integer, primary_key=True)
name = Column('name', String(50))
employees = relation("Person")
- class Person(Base, Fixture):
+ class Person(Base, ComparableEntity):
__tablename__ = 'people'
id = Column('id', Integer, primary_key=True)
- company_id = Column('company_id', Integer, ForeignKey('companies.id'))
+ company_id = Column('company_id', Integer,
+ ForeignKey('companies.id'))
name = Column('name', String(50))
discriminator = Column('type', String(50))
primary_language = Column('primary_language', String(50))
sess.flush()
sess.clear()
- self.assertEquals(sess.query(Person).filter(Engineer.primary_language=='cobol').first(), Engineer(name='vlad'))
- self.assertEquals(sess.query(Company).filter(Company.employees.of_type(Engineer).any(Engineer.primary_language=='cobol')).first(), c2)
+ eq_((sess.query(Person).
+ filter(Engineer.primary_language == 'cobol').first()),
+ Engineer(name='vlad'))
+ eq_((sess.query(Company).
+ filter(Company.employees.of_type(Engineer).
+ any(Engineer.primary_language == 'cobol')).first()),
+ c2)
def test_with_explicit_autoloaded(self):
meta = MetaData(testing.db)
- t1 = Table('t1', meta, Column('id', String(50), primary_key=True), Column('data', String(50)))
+ t1 = Table('t1', meta,
+ Column('id', String(50), primary_key=True),
+ Column('data', String(50)))
meta.create_all()
try:
class MyObj(Base):
sess.save(m)
sess.flush()
- assert t1.select().execute().fetchall() == [('someid', 'somedata')]
+ eq_(t1.select().execute().fetchall(), [('someid', 'somedata')])
finally:
meta.drop_all()
-class DeclarativeReflectionTest(TestBase):
+class DeclarativeReflectionTest(testing.TestBase):
def setUpAll(self):
global reflection_metadata
reflection_metadata = MetaData(testing.db)
def setUp(self):
global Base
- Base = declarative_base(testing.db)
+ Base = decl.declarative_base(testing.db)
def tearDown(self):
for t in reflection_metadata.table_iterator():
def test_basic(self):
meta = MetaData(testing.db)
- class User(Base, Fixture):
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
__autoload__ = True
addresses = relation("Address", backref="user")
- class Address(Base, Fixture):
+ class Address(Base, ComparableEntity):
__tablename__ = 'addresses'
__autoload__ = True
sess.flush()
sess.clear()
- self.assertEquals(sess.query(User).all(), [User(name='u1', addresses=[
+ eq_(sess.query(User).all(), [User(name='u1', addresses=[
Address(email='one'),
Address(email='two'),
])])
- a1 = sess.query(Address).filter(Address.email=='two').one()
- self.assertEquals(a1, Address(email='two'))
- self.assertEquals(a1.user, User(name='u1'))
+ a1 = sess.query(Address).filter(Address.email == 'two').one()
+ eq_(a1, Address(email='two'))
+ eq_(a1.user, User(name='u1'))
def test_rekey(self):
meta = MetaData(testing.db)
- class User(Base, Fixture):
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
__autoload__ = True
nom = Column('name', String(50), key='nom')
addresses = relation("Address", backref="user")
- class Address(Base, Fixture):
+ class Address(Base, ComparableEntity):
__tablename__ = 'addresses'
__autoload__ = True
sess.flush()
sess.clear()
- self.assertEquals(sess.query(User).all(), [User(nom='u1', addresses=[
+ eq_(sess.query(User).all(), [User(nom='u1', addresses=[
Address(email='one'),
Address(email='two'),
])])
- a1 = sess.query(Address).filter(Address.email=='two').one()
- self.assertEquals(a1, Address(email='two'))
- self.assertEquals(a1.user, User(nom='u1'))
+ a1 = sess.query(Address).filter(Address.email == 'two').one()
+ eq_(a1, Address(email='two'))
+ eq_(a1.user, User(nom='u1'))
self.assertRaises(TypeError, User, name='u3')
def test_supplied_fk(self):
meta = MetaData(testing.db)
- class IMHandle(Base, Fixture):
+ class IMHandle(Base, ComparableEntity):
__tablename__ = 'imhandles'
__autoload__ = True
user_id = Column('user_id', Integer,
ForeignKey('users.id'))
- class User(Base, Fixture):
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
__autoload__ = True
handles = relation("IMHandle", backref="user")
sess.flush()
sess.clear()
- self.assertEquals(sess.query(User).all(), [User(name='u1', handles=[
+ eq_(sess.query(User).all(), [User(name='u1', handles=[
IMHandle(network='blabber', handle='foo'),
IMHandle(network='lol', handle='zomg')
])])
- a1 = sess.query(IMHandle).filter(IMHandle.handle=='zomg').one()
- self.assertEquals(a1, IMHandle(network='lol', handle='zomg'))
- self.assertEquals(a1.user, User(name='u1'))
+ a1 = sess.query(IMHandle).filter(IMHandle.handle == 'zomg').one()
+ eq_(a1, IMHandle(network='lol', handle='zomg'))
+ eq_(a1.user, User(name='u1'))
def test_synonym_for(self):
- class User(Base, Fixture):
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True)
name = Column('name', String(50))
- @synonym_for('name')
+ @decl.synonym_for('name')
@property
def namesyn(self):
return self.name
sess = create_session()
u1 = User(name='someuser')
- assert u1.name == "someuser", u1.name
- assert u1.namesyn == 'someuser', u1.namesyn
+ eq_(u1.name, "someuser")
+ eq_(u1.namesyn, 'someuser')
sess.save(u1)
sess.flush()
- rt = sess.query(User).filter(User.namesyn=='someuser').one()
- self.assertEquals(rt, u1)
+ rt = sess.query(User).filter(User.namesyn == 'someuser').one()
+ eq_(rt, u1)
def test_comparable_using(self):
- class NameComparator(PropComparator):
+ class NameComparator(sa.orm.PropComparator):
@property
def upperself(self):
cls = self.prop.parent.class_
col = getattr(cls, 'name')
- return func.upper(col)
+ return sa.func.upper(col)
def operate(self, op, other, **kw):
return op(self.upperself, other, **kw)
- class User(Base, Fixture):
+ class User(Base, ComparableEntity):
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True)
name = Column('name', String(50))
- @comparable_using(NameComparator)
+ @decl.comparable_using(NameComparator)
@property
def uc_name(self):
return self.name is not None and self.name.upper() or None
sess = create_session()
u1 = User(name='someuser')
- assert u1.name == "someuser", u1.name
- assert u1.uc_name == 'SOMEUSER', u1.uc_name
+ eq_(u1.name, "someuser", u1.name)
+ eq_(u1.uc_name, 'SOMEUSER', u1.uc_name)
sess.save(u1)
sess.flush()
sess.clear()
- rt = sess.query(User).filter(User.uc_name=='SOMEUSER').one()
- self.assertEquals(rt, u1)
+ rt = sess.query(User).filter(User.uc_name == 'SOMEUSER').one()
+ eq_(rt, u1)
sess.clear()
rt = sess.query(User).filter(User.uc_name.startswith('SOMEUSE')).one()
- self.assertEquals(rt, u1)
+ eq_(rt, u1)
+
if __name__ == '__main__':
testing.main()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
