使用SQLAlchemy中的动态功能完全重新启动/重新加载声明性类

Question

我使用SQLAlchemy+SQLite3基于用户输入创建多个数据库。初始化新数据库时，用户定义任意数量的任意功能及其类型。我编写了一个DBManager类作为用户输入和数据库创建/访问之间的接口

在声明性模型（类Features）中动态地“注入”这些任意特性的工作与预期的一样。我遇到的问题是，当用户想要创建第二个/不同的数据库时：我不知道如何完全“清除”或“刷新”模型或declarative_base，以便用户能够创建一个新的数据库（可能具有不同的功能）

下面是我的情况的一个最小可复制示例：

src.__init__.py：

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Session = sessionmaker()
Base = declarative_base()

src.features.py

from sqlalchemy import Column, ForeignKey, Integer
from sqlalchemy.orm import relationship

from src import Base

class Features(Base):
    __tablename__ = "features"
    features_id = Column(Integer, primary_key=True)

    @classmethod
    def add_feature(cls, feature_name, feature_type):
        setattr(cls, feature_name, Column(feature_type))

src.db_manager.py：

from typing import Optional, Dict

from sqlalchemy import create_engine

from src import Base, Session
from src.features import Features


class DBManager:
    def __init__(self, path: str, features: Optional[Dict] = None) -> None:
        self.engine = create_engine(f'sqlite:///{path}')
        Session.configure(bind=self.engine)
        self.session = Session()
        self.features = features
        if self.features:  # user passed in some arbitrary features
            self.bind_features_to_features_table()
        Base.metadata.create_all(bind=self.engine)

    def bind_features_to_features_table(self):
        for feature_name, feature_type in self.features.items():
            Features.add_feature(feature_name=feature_name, feature_type=feature_type)

我希望能够做到这样：

from sqlalchemy import String, Float, Integer
from src.db_manager import DBManager

# User wants to create a database with these features
features = {
    'name': String,
    'height': Float,
}
db_manager = DBManager(path='my_database.db', features=features)

# ... User does some stuff with database here ...

# Now the user wants to create another database with these features
other_features = {
    'age': Integer,
    'weight': Float,
    'city_of_residence': String,
    'name': String,
}
db_manager = DBManager(path='another_database.db', features=other_features)

在执行最后一行之后，我遇到：InvalidRequestError: Implicitly combining column features.name with column features.name under attribute 'name'. Please configure one or more attributes for these same-named columns explicitly.如果功能name没有出现在两个数据库上，则不会发生错误，但是功能height将被带到第二个数据库，这是不需要的

我尝试过但没有成功的事情：

• 在DBManager实例之间调用Base.metadata.clear()：相同错误
• 在DBManager实例之间调用sqlalchemy.orm.clear_mappers()：结果为AttributeError: 'NoneType' object has no attribute 'instrument_attribute'
• 调用delattr(Features, feature_name)：结果为NotImplementedError: Can't un-map individual mapped attributes on a mapped class.

这个程序将在GUI中运行，因此我实在无法退出/重新启动脚本以连接到第二个数据库。用户应该能够加载/创建不同的数据库，而无需关闭程序

我知道错误源于底层Base对象尚未“刷新”，并且仍在跟踪在我的第一个DBManager实例中创建的功能。然而，我不知道如何解决这个问题。更糟糕的是，任何覆盖/重新加载新Base对象的尝试都需要应用于从__init__.py导入该对象的所有模块，这听起来很棘手。有人能解决这个问题吗

Answer 1

我的解决方案是在函数get_features内定义Features声明性类，该函数将Base（声明性基）实例作为参数。该函数返回Features类对象，因此每次调用基本上都会作为一个整体创建一个新的Features类

然后类DBManager负责调用该函数，并且Features成为DBManager的实例属性。创建DBManager的新实例意味着基于Features创建一个整个新类，然后我可以向其中添加我想要的任意特性

代码如下所示：

def get_features(declarative_base):
    class Features(declarative_base):
        __tablename__ = "features"
        features_id = Column(Integer, primary_key=True)

        @classmethod
        def add_feature(cls, feature_name, feature_type):
            setattr(cls, feature_name, Column(feature_type))

    return Features


class DBManager:
    def __init__(self, path, features):
        self.engine = create_engine(f'sqlite:///{path}')
        Session.configure(bind=self.engine)
        self.session = Session()
        base = declarative_base()
        self.features_table = get_features(base=base)
        if self.features:  # user passed in some arbitrary features
            self.bind_features_to_features_table()
        Base.metadata.create_all(bind=self.engine)

    def bind_features_to_features_table(self):
        for feature_name, feature_type in self.features.items():
            self.features_table.add_feature(feature_name=feature_name, feature_type=feature_type)

这确实让人觉得有点费解，我不知道是否有任何我不知道的警告，但据我所知，这种方法解决了我的问题