python - 无法在flask-python中导入动态生成的sqlalchemy类

Question

刚开始学习 Flask 和 Python，我必须创建 10 多个具有相同结构和约束的表(MySQLAlchemy)。因此，在寻找比单独声明每个类更快的解决方案之后，我在 models/user.py 中执行了以下操作:

#models/users.py
class exchangeapis(object):

email = db.Column(db.String(50), primary_key=True)
api =db.Column(db.String(100))
secret= db.Column(db.String(100))



@declared_attr.cascading 

# Makes sure all the dbs derived from this calss get the fK and PK

def email(cls):
        #this is necessary to propagate fk and pk to the instances
        return db.Column(db.String(50), db.ForeignKey('users.email'), primary_key=True)

#general methods to get class data       
def get_api(self):
    return self.api

def get_secret(self):
    return self.secret

exchange=['A','B','C']    
for exchange in exchanges:

cls=type(exchange.title(), (exchangeapis, db.Model), { '__tablename__' : str(exchange)+"_api"})
print(cls.__name__)

print(cls.__name__) 

给了我“A”，“B”，“C”，我希望是“A_api”，“B_api”...

当我尝试将其导入我的 account/views.py 文件时，包含以下内容:

from ..models import User, A_api

我得到“File “/home/puru991/flask-saas/flask-base/app/account/views.py”，第 8 行， 从..models导入用户，A_api

导入错误:无法导入名称“A_api””

但是如果我按以下方式定义该类:

class A_api(db.Model):
__tablename__="A_api"
email = db.Column(db.String(50),ForeignKey("users.email"), primary_key=True)
api =db.Column(db.String(100))
secret= db.Column(db.String(100))

#general methods to get class data       
def get_api(self):
    return self.api

def get_secret(self):
    return self.secret

没有错误。所以我的问题是如何基于exchangeapi(object)动态创建类并成功导入它们？我在这里浏览了大约 8-10 个答案，我最多能学到的是如何动态创建它们。我还了解到 MySQLAlchemy 创建与表名同名的类，所以我认为通过表名(即“A_api”)引用就可以了。我错过了什么？

Answer 1

您无法导入它们的原因是您的 users.py 文件中没有名为 A_api 的变量。

exchange=['A','B','C']    
for exchange in exchanges:
    cls=type(exchange.title(), (exchangeapis, db.Model), { '__tablename__' : str(exchange)+"_api"})
    print(cls.__name__)

在上面的代码中，您创建了三个类，每次将这些类分配给名为 cls 的变量。一旦 for 循环完成，您的变量 cls 将是最后创建的类，从技术上讲，您可以将此变量导入到另一个模块中(但不要这样做)。

如果运行以下代码，您可以看到这一点:

exchange=['A','B','C']    
for exchange in exchanges:
    cls=type(exchange.title(), (exchangeapis, db.Model), { '__tablename__' : str(exchange)+"_api"})

# This will print C_api as it's the last assignment to cls in the for loop
print(cls.__name__)

关于表名/类名；你搞错了，Flask-SQLAlchemy derives the table name从类名中，将名为“CamelCase”的类转换为名为“camel_case”的表，除非您使用 __tablename__ 覆盖表名。允许定义表名允许您使用更Pythonic的命名类名，但使用更传统的数据库表名。

请参阅下面的工作代码的简单演示，两个文件 models.py 和 run.py:

models.py

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

# This list will hold the dynamically created classes
dynamic_models = []

db = SQLAlchemy()


class User(db.Model):
    __tablename__ = 'users'

    email = db.Column(db.String(254), primary_key=True)
    # Other User fields


class ExchangeApiMixin(object):

    @declared_attr.cascading
    def email(cls):
        return db.Column(db.String(254), db.ForeignKey('users.email'), primary_key=True)

    api = db.Column(db.String(100))

    secret = db.Column(db.String(100))


for exchange in ['A_api', 'B_api', 'C_api']:
    cls = type(exchange, (ExchangeApiMixin, db.Model), {})
    print(cls.__name__)
    dynamic_models.append(cls)

# This will print C_api as it's the last assignment to cls in the for loop
print(cls.__name__)


# Individual variables also reference the dynamically created classes. The could be any legal python variable name
A_API = dynamic_models[0]
B_API = dynamic_models[1]
C_API = dynamic_models[2]

# Succinct way of creating the dynamic classes
(D_API, E_API, F_API) = [type(exchange, (ExchangeApiMixin, db.Model), {}) for exchange in ['D_api', 'E_api', 'F_api']]

run.py

import random, string
from flask import Flask, render_template_string
from models import db, dynamic_models, A_API, B_API, C_API, cls, D_API

app = Flask(__name__)
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
db.init_app(app)


_template = '''
    {% for rows in queries %}
        <table border='1'>
            <thead>
            <tr>
            <th>Class</th>
            <th>Email</th>
            <th>API</th>
            <th>Secret</th>
            </tr>
            </thead>    
            {% for row in rows %}
                <tr>
                <td>{{row.__class__.__name__}}</td>
                <td>{{row.email}}</td>
                <td>{{row.api}}</td>
                <td>{{row.secret}}</td>
                </tr>
            {% endfor %}
        </table>
        <p></p>
    {% endfor %}        
'''


@app.route('/')
def index():
    # display all the A_API, B_API, C_API, D_API instances
    a_rows = A_API.query.all()
    b_rows = B_API.query.all()
    c_rows = C_API.query.all()
    d_rows = D_API.query.all()
    return render_template_string(_template, queries=[a_rows, b_rows, c_rows, d_rows])


def build_sample_db():
    with app.app_context():

        db.drop_all()
        db.create_all()

        def create_from_model(Model, api_name):
            for _ in range(0, 10):
                _model = Model(
                    email='{mailbox}@example.com'.format(mailbox=''.join(random.choices(string.ascii_lowercase + string.digits, k=10))),
                    api=api_name,
                    secret='pa$$w0rd'
                )
                db.session.add(_model)

            db.session.commit()

        # Create instances using A_API, could pass dynamic_models[0]
        create_from_model(Model=A_API, api_name='A API Name')

        # Create using B_API instances
        create_from_model(Model=dynamic_models[1], api_name='B API Name')

        # Create using C_API instances using the cls variable (which is the 'C' class
        create_from_model(Model=cls, api_name='C API Name')

        # Create using D_API instances
        create_from_model(Model=D_API, api_name='D API Name')


if __name__ == '__main__':
    build_sample_db()
    app.run()