如何修复我在普通 SQLAlchemy 和 Flask-SQLAlchemy 中使用相同模型的方法？

Question

我cameacross several 关于如何在 Flask-SQLAlchemy 中使用 vanilla SQLAlchemy 模型的方法。

在 Flask-SQLAlchemy 中使用继承自 Base 的模型就像一种魅力。

但我真的很喜欢那种方便的东西......

Job.query.all() # Does not work
db.session.query(Job).all() # Works

所以我开始着手处理这个问题并整理了一些代码，但我被困住了，需要一些帮助才能让它变得干净整洁。

以下块是一个通用定义，不继承自其中任何一个。 它是从 Flask-SQLAlchemy 和 vanilla SQLAlchemy 导入并在某些时候使用的。

class VanillaMachine():

    __tablename__ = 'machine'

    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    status = Column(Integer)

还有一个工厂接受 db.Model 或 Base 并返回 Machine 和正确的父级：

class MachineFactory:

    def __init__(self, *args, **kwargs):
        pass

    def __new__(cls, *args, **kwargs):

        return type('Machine',(object, VanillaMachine, args[0]), VanillaMachine.__dict__.copy())

我很确定该代码有问题，但是 我不确定在哪里。

如果我像这样使用它

db = SQLAlchemy()

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()#

Machine1 = MachineFactory(db.Model)
Machine2 = MachineFactory(Base)

有错误提示

sqlalchemy.exc.ArgumentError: Column object 'id' already assigned to Table 'machine'

可以帮助我以一种好的、可靠的方式解决这个问题吗？

我知道你可以只使用一个函数，将 parent 作为参数传递给 VanillaMachine 并使用一些 if 语句，但这太简单了，对吧？ :)

编辑：

我遇到的其他方法是

1. 使用 Flask 上下文来使用 Flask-SQLAlchemy 模型

   with app.app_context():
       pass
   
   or 
   
   app.app_context().push()
   

但这对我来说太专注于 Flask 并且不允许清楚地分离模型，使它们独立并适应上下文。

1. 为db = SQLAlchemy(app, model_class=Base) 提供替代基类，请参阅here。这可能对我有用，但到目前为止我还没有对此进行评估。

Answer 1

我找到了一个受Factory pattern 启发的好解决方案，并且 Declarative Mixins 如 SQLAlchemy 文档中所述。

对于复杂的多级继承场景，需要different approach，使用@declared_attr.cascading。

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, String
from sqlalchemy import MetaData

from sqlalchemy.ext.declarative import declarative_base
from flask_sqlalchemy import SQLAlchemy

SQLALCHEMY_DATABASE_URI = 'sqlite:///' + '/tmp/test_app.db'
engine = create_engine(SQLALCHEMY_DATABASE_URI, echo=True)

# for vanilla
Base = declarative_base()

# for Flask (import from app once initialized)
db = SQLAlchemy()


class MachineMixin:

    __tablename__ = 'machine'
    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    status = Column(Integer)


class ModelFactory:

    @staticmethod
    def create(which_model, which_parent):

        if which_parent == 'flask_sqlalchemy':

            parent = db.Model

        elif which_parent == 'pure_sqlalchemy':

            parent = Base

        # now use type() to interit, fill __dict__ and assign a name
        obj = type(which_model.__name__ + '_' + which_parent,
                    (which_model, parent),
                    {})
        return obj


test_scenario = 'pure_sqlalchemy' # 'flask_sqlalchemy'

Machine = ModelFactory.create(MachineMixin, test_scenario)

if test_scenario == 'flask_sqlalchemy':

    db.metadata.drop_all(bind=engine)
    db.metadata.create_all(bind=engine)

elif test_scenario == 'pure_sqlalchemy':

    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)


Session = sessionmaker(bind=engine)
session = Session()
session.add(Machine(name='Bob', status=1))
session.commit()