of a join in place of the "of_type()" modifier.
[ticket:2333]
+ - [feature] The "deferred declarative
+ reflection" system has been moved into the
+ declarative extension itself, using the
+ new DeferredReflection class. This
+ class is now tested with both single
+ and joined table inheritance use cases.
+ [ticket:2485]
+
- [feature] Added new core function "inspect()",
which serves as a generic gateway to
introspection into mappers, objects,
deprecated; use Inspector.get_pk_constraint().
Courtesy Diana Clarke. [ticket:2422]
+ - [bug] The autoload_replace flag on Table,
+ when False, will cause any reflected foreign key
+ constraints which refer to already-declared
+ columns to be skipped, assuming that the
+ in-Python declared column will take over
+ the task of specifying in-Python ForeignKey
+ or ForeignKeyConstraint declarations.
+
- sql
- [feature] The Inspector object can now be
acquired using the new inspect() service,
.. automodule:: beaker_caching
-.. _examples_declarative_reflection:
-
-Declarative Reflection
-----------------------
-
-Location: /examples/declarative_reflection
-
-.. automodule:: declarative_reflection
-
Directed Graphs
---------------
.. autoclass:: ConcreteBase
+.. autoclass:: DeferredReflection
+++ /dev/null
-"""
-Illustrates how to mix table reflection with Declarative, such that
-the reflection process itself can take place **after** all classes
-are defined. Declarative classes can also override column
-definitions loaded from the database.
-
-At the core of this example is the ability to change how Declarative
-assigns mappings to classes. The ``__mapper_cls__`` special attribute
-is overridden to provide a function that gathers mapping requirements
-as they are established, without actually creating the mapping.
-Then, a second class-level method ``prepare()`` is used to iterate
-through all mapping configurations collected, reflect the tables
-named within and generate the actual mappers.
-
-The example is new in 0.7.5 and makes usage of the new
-``autoload_replace`` flag on :class:`.Table` to allow declared
-classes to override reflected columns.
-
-Usage example::
-
- Base = declarative_base(cls=DeclarativeReflectedBase)
-
- class Foo(Base):
- __tablename__ = 'foo'
- bars = relationship("Bar")
-
- class Bar(Base):
- __tablename__ = 'bar'
-
- # illustrate overriding of "bar.foo_id" to have
- # a foreign key constraint otherwise not
- # reflected, such as when using MySQL
- foo_id = Column(Integer, ForeignKey('foo.id'))
-
- Base.prepare(e)
-
- s = Session(e)
-
- s.add_all([
- Foo(bars=[Bar(data='b1'), Bar(data='b2')], data='f1'),
- Foo(bars=[Bar(data='b3'), Bar(data='b4')], data='f2')
- ])
- s.commit()
-
-
-"""
+++ /dev/null
-from sqlalchemy import *
-from sqlalchemy.orm import *
-from sqlalchemy.orm.util import _is_mapped_class
-from sqlalchemy.ext.declarative import declarative_base, declared_attr
-
-class DeclarativeReflectedBase(object):
- _mapper_args = []
-
- @classmethod
- def __mapper_cls__(cls, *args, **kw):
- """Declarative will use this function in lieu of
- calling mapper() directly.
-
- Collect each series of arguments and invoke
- them when prepare() is called.
- """
-
- cls._mapper_args.append((args, kw))
-
- @classmethod
- def prepare(cls, engine):
- """Reflect all the tables and map !"""
- while cls._mapper_args:
- args, kw = cls._mapper_args.pop()
- klass = args[0]
- # autoload Table, which is already
- # present in the metadata. This
- # will fill in db-loaded columns
- # into the existing Table object.
- if args[1] is not None:
- table = args[1]
- Table(table.name,
- cls.metadata,
- extend_existing=True,
- autoload_replace=False,
- autoload=True,
- autoload_with=engine,
- schema=table.schema)
-
- # see if we need 'inherits' in the
- # mapper args. Declarative will have
- # skipped this since mappings weren't
- # available yet.
- for c in klass.__bases__:
- if _is_mapped_class(c):
- kw['inherits'] = c
- break
-
- klass.__mapper__ = mapper(*args, **kw)
-
-if __name__ == '__main__':
- Base = declarative_base()
-
- # create a separate base so that we can
- # define a subset of classes as "Reflected",
- # instead of everything.
- class Reflected(DeclarativeReflectedBase, Base):
- __abstract__ = True
-
- class Foo(Reflected):
- __tablename__ = 'foo'
- bars = relationship("Bar")
-
- class Bar(Reflected):
- __tablename__ = 'bar'
-
- # illustrate overriding of "bar.foo_id" to have
- # a foreign key constraint otherwise not
- # reflected, such as when using MySQL
- foo_id = Column(Integer, ForeignKey('foo.id'))
-
- e = create_engine('sqlite://', echo=True)
- e.execute("""
- create table foo(
- id integer primary key,
- data varchar(30)
- )
- """)
-
- e.execute("""
- create table bar(
- id integer primary key,
- data varchar(30),
- foo_id integer
- )
- """)
-
- Reflected.prepare(e)
-
- s = Session(e)
-
- s.add_all([
- Foo(bars=[Bar(data='b1'), Bar(data='b2')], data='f1'),
- Foo(bars=[Bar(data='b3'), Bar(data='b4')], data='f2')
- ])
- s.commit()
- for f in s.query(Foo):
- print f.data, ",".join([b.data for b in f.bars])
\ No newline at end of file
for fkey_d in fkeys:
conname = fkey_d['name']
constrained_columns = fkey_d['constrained_columns']
+ if exclude_columns and set(constrained_columns).intersection(
+ exclude_columns):
+ continue
referred_schema = fkey_d['referred_schema']
referred_table = fkey_d['referred_table']
referred_columns = fkey_d['referred_columns']
However, one improvement that can be made here is to not
require the :class:`.Engine` to be available when classes are
-being first declared. To achieve this, use the example
-described at :ref:`examples_declarative_reflection` to build a
-declarative base that sets up mappings only after a special
-``prepare(engine)`` step is called::
+being first declared. To achieve this, use the
+:class:`.DeferredReflection` mixin, which sets up mappings
+only after a special ``prepare(engine)`` step is called::
- Base = declarative_base(cls=DeclarativeReflectedBase)
+ from sqlalchemy.ext.declarative import declarative_base, DeferredReflection
+
+ Base = declarative_base(cls=DeferredReflection)
class Foo(Base):
__tablename__ = 'foo'
Base.prepare(e)
-
+.. versionadded:: 0.8
+ Added :class:`.DeferredReflection`.
+
Mapper Configuration
====================
def prepare(cls, engine):
"""Reflect all :class:`.Table` objects for all current
:class:`.DeferredReflection` subclasses"""
- to_map = set([m for m in _MapperConfig.configs.values()
- if issubclass(m.cls, cls)])
+ to_map = [m for m in _MapperConfig.configs.values()
+ if issubclass(m.cls, cls)]
for thingy in to_map:
cls.__prepare__(thingy.args, engine)
thingy.map()
assert len(t2.indexes) == 2
@testing.provide_metadata
- def test_autoload_replace_foreign_key(self):
+ def test_autoload_replace_foreign_key_nonpresent(self):
+ """test autoload_replace=False with col plus FK
+ establishes the FK not present in the DB.
+
+ """
a = Table('a', self.metadata, Column('id', Integer, primary_key=True))
b = Table('b', self.metadata, Column('id', Integer, primary_key=True),
Column('a_id', Integer))
assert b2.c.a_id.references(a2.c.id)
eq_(len(b2.constraints), 2)
+ @testing.provide_metadata
+ def test_autoload_replace_foreign_key_ispresent(self):
+ """test autoload_replace=False with col plus FK mirroring
+ DB-reflected FK skips the reflected FK and installs
+ the in-python one only.
+
+ """
+ a = Table('a', self.metadata, Column('id', Integer, primary_key=True))
+ b = Table('b', self.metadata, Column('id', Integer, primary_key=True),
+ Column('a_id', Integer, sa.ForeignKey('a.id')))
+ self.metadata.create_all()
+
+ m2 = MetaData()
+ b2 = Table('b', m2, Column('a_id', Integer, sa.ForeignKey('a.id')))
+ a2 = Table('a', m2, autoload=True, autoload_with=testing.db)
+ b2 = Table('b', m2, extend_existing=True, autoload=True,
+ autoload_with=testing.db,
+ autoload_replace=False)
+
+ assert b2.c.id is not None
+ assert b2.c.a_id.references(a2.c.id)
+ eq_(len(b2.constraints), 2)
+
+ @testing.provide_metadata
+ def test_autoload_replace_foreign_key_removed(self):
+ """test autoload_replace=False with col minus FK that's in the
+ DB means the FK is skipped and doesn't get installed at all.
+
+ """
+ a = Table('a', self.metadata, Column('id', Integer, primary_key=True))
+ b = Table('b', self.metadata, Column('id', Integer, primary_key=True),
+ Column('a_id', Integer, sa.ForeignKey('a.id')))
+ self.metadata.create_all()
+
+ m2 = MetaData()
+ b2 = Table('b', m2, Column('a_id', Integer))
+ a2 = Table('a', m2, autoload=True, autoload_with=testing.db)
+ b2 = Table('b', m2, extend_existing=True, autoload=True,
+ autoload_with=testing.db,
+ autoload_replace=False)
+
+ assert b2.c.id is not None
+ assert not b2.c.a_id.references(a2.c.id)
+ eq_(len(b2.constraints), 1)
+
@testing.provide_metadata
def test_autoload_replace_primary_key(self):
a = Table('a', self.metadata, Column('id', Integer))
global Base
Base = decl.declarative_base(testing.db)
+ def teardown(self):
+ super(DeclarativeReflectionBase, self).teardown()
+ clear_mappers()
+
class DeclarativeReflectionTest(DeclarativeReflectionBase):
@classmethod
DefBase.prepare(testing.db)
self._roundtrip()
+
+ def test_redefine_fk_double(self):
+ class User(decl.DeferredReflection, fixtures.ComparableEntity,
+ Base):
+ __tablename__ = 'users'
+ addresses = relationship("Address", backref="user")
+
+ class Address(decl.DeferredReflection, fixtures.ComparableEntity,
+ Base):
+ __tablename__ = 'addresses'
+ user_id = Column(Integer, ForeignKey('users.id'))
+
+ decl.DeferredReflection.prepare(testing.db)
+ self._roundtrip()
+
+
+class DeferredInhReflectBase(DeferredReflectBase):
+ def _roundtrip(self):
+ Foo = Base._decl_class_registry['Foo']
+ Bar = Base._decl_class_registry['Bar']
+
+ s = Session(testing.db)
+
+ s.add_all([
+ Bar(data='d1', bar_data='b1'),
+ Bar(data='d2', bar_data='b2'),
+ Bar(data='d3', bar_data='b3'),
+ Foo(data='d4')
+ ])
+ s.commit()
+
+ eq_(
+ s.query(Foo).order_by(Foo.id).all(),
+ [
+ Bar(data='d1', bar_data='b1'),
+ Bar(data='d2', bar_data='b2'),
+ Bar(data='d3', bar_data='b3'),
+ Foo(data='d4')
+ ]
+ )
+
+class DeferredSingleInhReflectionTest(DeferredInhReflectBase):
+ @classmethod
+ def define_tables(cls, metadata):
+ Table("foo", metadata,
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('type', String(32)),
+ Column('data', String(30)),
+ Column('bar_data', String(30))
+ )
+
+ def test_basic(self):
+ class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
+ Base):
+ __tablename__ = 'foo'
+ __mapper_args__ = {"polymorphic_on":"type",
+ "polymorphic_identity":"foo"}
+
+ class Bar(Foo):
+ __mapper_args__ = {"polymorphic_identity":"bar"}
+
+ decl.DeferredReflection.prepare(testing.db)
+ self._roundtrip()
+
+ def test_add_subclass_column(self):
+ class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
+ Base):
+ __tablename__ = 'foo'
+ __mapper_args__ = {"polymorphic_on":"type",
+ "polymorphic_identity":"foo"}
+
+ class Bar(Foo):
+ __mapper_args__ = {"polymorphic_identity":"bar"}
+ bar_data = Column(String(30))
+
+ decl.DeferredReflection.prepare(testing.db)
+ self._roundtrip()
+
+ def test_add_pk_column(self):
+ class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
+ Base):
+ __tablename__ = 'foo'
+ __mapper_args__ = {"polymorphic_on":"type",
+ "polymorphic_identity":"foo"}
+ id = Column(Integer, primary_key=True)
+
+ class Bar(Foo):
+ __mapper_args__ = {"polymorphic_identity":"bar"}
+
+ decl.DeferredReflection.prepare(testing.db)
+ self._roundtrip()
+
+class DeferredJoinedInhReflectionTest(DeferredInhReflectBase):
+ @classmethod
+ def define_tables(cls, metadata):
+ Table("foo", metadata,
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('type', String(32)),
+ Column('data', String(30)),
+ )
+ Table('bar', metadata,
+ Column('id', Integer, ForeignKey('foo.id'), primary_key=True),
+ Column('bar_data', String(30))
+ )
+
+ def test_basic(self):
+ class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
+ Base):
+ __tablename__ = 'foo'
+ __mapper_args__ = {"polymorphic_on":"type",
+ "polymorphic_identity":"foo"}
+
+ class Bar(Foo):
+ __tablename__ = 'bar'
+ __mapper_args__ = {"polymorphic_identity":"bar"}
+
+ decl.DeferredReflection.prepare(testing.db)
+ self._roundtrip()
+
+ def test_add_subclass_column(self):
+ class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
+ Base):
+ __tablename__ = 'foo'
+ __mapper_args__ = {"polymorphic_on":"type",
+ "polymorphic_identity":"foo"}
+
+ class Bar(Foo):
+ __tablename__ = 'bar'
+ __mapper_args__ = {"polymorphic_identity":"bar"}
+ bar_data = Column(String(30))
+
+ decl.DeferredReflection.prepare(testing.db)
+ self._roundtrip()
+
+ def test_add_pk_column(self):
+ class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
+ Base):
+ __tablename__ = 'foo'
+ __mapper_args__ = {"polymorphic_on":"type",
+ "polymorphic_identity":"foo"}
+ id = Column(Integer, primary_key=True)
+
+ class Bar(Foo):
+ __tablename__ = 'bar'
+ __mapper_args__ = {"polymorphic_identity":"bar"}
+
+ decl.DeferredReflection.prepare(testing.db)
+ self._roundtrip()
+
+ def test_add_fk_pk_column(self):
+ class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
+ Base):
+ __tablename__ = 'foo'
+ __mapper_args__ = {"polymorphic_on":"type",
+ "polymorphic_identity":"foo"}
+
+ class Bar(Foo):
+ __tablename__ = 'bar'
+ __mapper_args__ = {"polymorphic_identity":"bar"}
+ id = Column(Integer, ForeignKey('foo.id'), primary_key=True)
+
+ decl.DeferredReflection.prepare(testing.db)
+ self._roundtrip()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
