Is it possible to unload declarative classes in SQLAlchemy?

Question

I’m working on a library where the user shall be able to simply declare a few classes which are automatically backed by the database. In short, somewhere hidden in the code, there is

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class LibraryBase(Base):
    # important library stuff

and the user should then do

class MyStuff(LibraryBase):
    # important personal stuff

class MyStuff_2(LibraryBase):
    # important personal stuff

mystuff = MyStuff()
Library.register(mystuff)
mystuff.changeIt() # apply some changes to the instance
Library.save(mystuff) # and save it

# same for all other classes

In a static environment, e.g. the user has created one file with all personal classes and imports this file, this works pretty well. All class names are fixed and SQLAlchemy knows how to map each class.

In an interactive environment, things are different: Now, there is a chance of a class being defined twice. Both classes might have different modules; but still SQLAlchemy will complain:

SAWarning: The classname 'MyStuff' is already in the registry of this declarative base, mapped to < class 'OtherModule.MyStuff' >

Is there a way to deal with this? Can I somehow unload a class from its declarative_base so that I can exchange its definition with a new one?

Answer 1

In my project I use this solution. Where library specified columns defined as mixin by declared_attr and target mapper created by type call with bases, as result I have full functional mapper.

from sqlalchemy import create_engine, BigInteger, Column
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.declarative import declared_attr


Base = declarative_base()


class LibraryBase(object):
    __tablename__ = 'model'

    @declared_attr
    def library_field(self):
        return Column(BigInteger)


class MyLibrary(object):

    @classmethod
    def register(cls, entity):
        tablename = entity.__tablename__
        Mapper = type('Entity_%s' % tablename, (Base, LibraryBase, entity), {
            '__tablename__': tablename,
            'id': Column(BigInteger, primary_key=True),
        })
        return Mapper

    @classmethod
    def setup(cls):
        Base.metadata.create_all()


class MyStaff(object):
    __tablename__ = 'sometable1'

    @declared_attr
    def staff_field(self):
        return Column(BigInteger)

    def mymethod(self):
        print('My method:', self)


class MyStaff2(MyStaff):
    __tablename__ = 'sometable2'


if __name__ == '__main__':
    engine = create_engine('sqlite://', echo=True)
    Base.metadata.bind = engine
    Session = scoped_session(sessionmaker(bind=engine))
    session = Session()

    # register and install
    MyStaffMapper = MyLibrary.register(MyStaff)
    MyStaffMapper2 = MyLibrary.register(MyStaff2)
    MyLibrary.setup()

    MyStaffMapper().mymethod()
    MyStaffMapper2().mymethod()

    session.query(MyStaffMapper.library_field) \
        .filter(MyStaffMapper.staff_field != None) \
        .all()