从SQLAlchemy模式实例获取ORM类

Question

在SQLAlchemy中，你可以这样进行查询：

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'

当然，我可以在save_model方法中添加model_class: Type[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