new reflection features/behaviors

Allow a Column to be passed to Table that already has that Table as its parent. It will be re-attached.
Don't blow away the FK on a Column when a new Column without that FK is coming in.
Don't blow away anything with the FK if the Column coming in is self
add exclude_columns argument to reflection functions.
add an "autoload_replace=False" flag so that we can control if autoload replaces existing cols or not when extend existing is turned on. I know this is more verbose but this grants the most flexibility.

with the autoload_replace=False feature, the example at UsageRecipes/DeclarativeReflectedBase can be simplified and enhanced so that columns can be declared normally on declared classes, and the table can then be "filled in" by reflection.

from sqlalchemy import *
from sqlalchemy.orm import *
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 !"""
        for args, kw in cls._mapper_args:
            klass = args[0](]

)
            klass.__table__ = table = Table(
                                        klass.__tablename__, 
                                        cls.metadata, 
                                        extend_existing=True,
                                        autoload_replace=False,
                                        autoload=True, 
                                        autoload_with=engine,
                                        )
            klass.__mapper__ = mapper(klass, table, **kw)


if __name__ == '__main__':
    Base= declarative_base(cls=DeclarativeReflectedBase)

    class Foo(Base):
        __tablename__ = 'foo'
        bars = relationship("Bar")

    class Bar(Base):
        __tablename__ = 'bar'
        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
    )
    """)

    Base.prepare(e)

    s = Session(e)

    s.add_all([       Foo(bars=[Bar(data='b1'), Bar(data='b2')](
), data='f1'),
        Foo(bars=[Bar(data='b4')](Bar(data='b3'),), data='f2')
    ])
    s.commit()
    for f in s.query(Foo):
        print f.data, ",".join([for b in f.bars](b.data))

test autoload_replace by itself:

from sqlalchemy import *

e = create_engine("sqlite://", echo=True)
m = MetaData()

b = Table('b',
    m,
    Column('a_id', Integer, ForeignKey('a.id'))
)

e.execute("create table a(id integer primary key)")
e.execute("create table b(id integer primary key, a_id integer)")

a = Table('a', m, autoload=True, autoload_with=e)
b = Table('b', m, extend_existing=True, autoload=True, autoload_with=e, autoload_replace=False)

assert b.c.id is not None
assert b.c.a_id.references(a.c.id)
assert len(b.constraints) == 2

test column replacement with itself:

from sqlalchemy import *

e = create_engine("sqlite://", echo=True)
m = MetaData()

a = Table('a', m, Column('id', Integer, primary_key=True))

aid = Column('a_id', ForeignKey('a.id'))
b = Table('b', m, aid)
b.append_column(aid)

assert b.c.a_id.references(a.c.id)
print b.constraints
assert len(b.constraints) == 2, len(b.constraints)

Comments (3)