在SQLAlchemy中可以卸载声明性类吗?

11 投票
3 回答
4170 浏览
提问于 2025-04-16 12:57

我正在开发一个库,用户可以简单地声明几个类,这些类会自动与数据库关联。简单来说,代码的某个地方隐藏着

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

然后用户应该这样做

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

在一个静态环境中,比如用户创建了一个包含所有个人类的文件并导入这个文件,这样的做法效果很好。所有的类名都是固定的,SQLAlchemy知道如何将每个类映射到数据库。

但是在一个交互式环境中,情况就不一样了:这时可能会出现同一个类被定义两次的情况。虽然这两个类可能来自不同的模块,但SQLAlchemy还是会报错:

SAWarning: 类名 'MyStuff' 已经在这个声明基础的注册表中,映射到 < class 'OtherModule.MyStuff' >

有没有办法解决这个问题?我能不能以某种方式将一个类从它的 declarative_base 中“卸载”,这样我就可以用新的定义替换掉它?

3 个回答

1

在我的项目中,我使用了这个解决方案。这里的库指定的列是通过 declared_attr 定义为混合属性的,而目标映射器是通过 type 调用并传入基础类创建的,最终我得到了一个功能齐全的映射器。

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() 
2

看起来,我不太确定这个方法是否有效,但我觉得你想要的是

sqlalchemy.orm.instrumentation.unregister_class()

http://hg.sqlalchemy.org/sqlalchemy/file/762548ff8eef/lib/sqlalchemy/orm/instrumentation.py#l466

4

你可以使用:

sqlalchemy.orm.instrumentation.unregister_class(cl)
del cl._decl_class_registry[cl.__name__]

第一行是为了防止你不小心使用了一个没有注册的类。第二行是用来注销的,这样就不会出现警告了。

撰写回答