从 SQL alchemy 中的模式实例获取 ORM 类

Question

在 SQL alchemy 中，你可以这样做来进行查询：

model_instance = session.query(SomeModel).filter(SomeModel.value == some_value).first()

其中

model_instance

相当于某种形式，省略数据：

model_instance = SomeModel( ... )

假设我想创建一个函数，它可以为

model_instance

采用多种可能的类型，每个类型都有自己的表。为此，我想做类似

type(model_instance)

的事情，除了获取一个 Python 类之外，我想要一个具有属性

'_sa_instance_state'

的对象，这样我就可以做

session.query(SomeModel)

，给定一个实例

SomeModel()

。

为了澄清起见，这里有一个示例方法（

save_model

）：

class Database:
    """SQLAlchemy object for interacting with the database"""

    def __init__(self, db_url: str):
        self.engine = create_engine(db_url)
        Base.metadata.create_all(self.engine)
        Session = sessionmaker(bind=self.engine)
        self.session = Session()
    
    def save_model(self, model: Model) -> bool:
        """Saves given model to the database"""
        try:
            model_type = ... # Get the ORM class object from the model here
            if self.contains(model_type, **model.__dict__):
                return False
            self.session.add(model)
            self.session.commit()
            return True
        except Exception as e:
            self.session.rollback()
            raise Exception(f"Unable to save model: {str(e)}")
        return False

    def contains(self, model_class: Type[Model], **kwargs) -> bool:
        """Returns True if a model of the specified class with the given attributes exists in the database."""
        query = self.session.query(model_class)
        for key, value in kwargs.items():
            query = query.filter(getattr(model_class, key) == value)
        return query.first() is not None

其中

Model

是模式类型的联合，每个模式类型都有自己单独的表。

我尝试做这样的事情

model_type = sqlalchemy.Table(model.__table__.name, model.__table__.metadata, autoload_with=self.engine)

和

model_type = sqlalchemy.sql.text(model.__table__.name)

和

model_type = type(model)

但我仍然遇到以下问题的一些变体：

'Table' object has no attribute '_sa_instance_state'

或

'TextClause' object has no attribute '_sa_instance_state'

或

'SomeModel' has no attribute '_sa_instance_state'

当然，我可以将

model_class: Type[Model]

作为另一个参数添加到

save_model

方法中，但我想知道是否可以避免这样做。

注意：如果您指定特定模型类型，则

contains

有效（例如，

contains(SomeModel, param1="some_value")

，但如果您指定

contains(type(model), param1="some_value")

则无效）。

Answer 1

实例的类可以通过以下任一方式获取：

sqlalchemy.inspect(some_model).mapper.class_

some_model.__class__

type(some_model)

但是问题是，在

contains

中，代码迭代实例的

__dict__

中的项目来创建过滤器，但实例具有模型上不存在的

_sa_instance_state

属性，因此

getattr(ModelClass, '_sa_instance_state')

会引发一个例外。

一个简单的解决方案是跳过此属性：

for k, v in kwargs.items():
    if k = 'sa_instance_state':
        continue
    # Build the filter

更复杂的方法可能是在尝试创建过滤器之前检查模型上是否存在该属性：

mapper = sqlalchemy.inspect(model_class)
for k, v kwargs.items():
    if k not in mapper.columns:
        continue
    # Build filter