Flask+SQLAlchemy搭建GraphQL时模型反射失败问题求助

我想用GraphQL搭配SQLAlchemy操作一个已建好表和关系（无数据）的PostgreSQL数据库，数据库连接已经成功。因为数据库schema是在Python外部创建的，所以用了automap_base。但把database.py导入app.py后，就算初始化了数据库，类变量还是None，导致GraphQL schema传的是“None”类型而不是“SQLAlchemyObjectType”。我之前做JavaScript开发，怀疑是不是类似异步问题，需要延迟调用getter，但不确定。

app.py 相关代码 #

from flask_graphql import GraphQLView
from database import init_db
from database import schema
...
init_db(app)

# 添加GraphQL端点
app.add_url_rule(
    "/graphql",
    view_func=GraphQLView.as_view(
    "graphql",
        schema=schema,
        graphiql=True,
    )
)

database.py 相关代码 #

# SQLAlchemy
db = SQLAlchemy()
Base = automap_base()

def init_db(app):
    global Base, db
    db.init_app(app)
    Base.prepare(db.engine, reflect=True)

# 获取模型
def get_user_model():
    return Base.classes.AppUser

def get_link_model():
    return Base.classes.Link

def get_link_stack_model():
    return Base.classes.LinkStack

def get_password_model():
    return Base.classes.Password

报错信息 #

File "/Users/vincenzomeschi/Documents/ListStack--V/server/database.py", line 35, in Meta
model = get_user_model()
^^^^^^^^^^^^^^^^   File "/Users/vincenzomeschi/Documents/ListStack--V/server/database.py", line 18, in get_user_model
return Base.classes.AppUser
^^^^^^^^^^^^^^^^^^^^   File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/sqlalchemy/util/_collections.py", line 188, in __getattr__
raise AttributeError(key) AttributeError: AppUser

问题核心是GraphQL Schema在Base.prepare()执行前就被加载了——你导入schema的时候，get_user_model()就被调用了，但此时Base还没完成数据库表的反射，自然找不到AppUser类。

解决步骤 #

1. 延迟GraphQL Schema的创建，不要在模块导入时就定义好，而是等到init_db执行完成后再生成。
2. 调整代码结构，把Schema的定义改成可延迟初始化的函数。

修改后的 database.py 示例

# SQLAlchemy
db = SQLAlchemy()
Base = automap_base()

def init_db(app):
    global Base, db
    db.init_app(app)
    Base.prepare(db.engine, reflect=True)
    # 初始化GraphQL Schema（如果Schema定义在database.py里）
    init_graphql_schema()

# 获取模型
def get_user_model():
    return Base.classes.AppUser

def get_link_model():
    return Base.classes.Link

def get_link_stack_model():
    return Base.classes.LinkStack

def get_password_model():
    return Base.classes.Password

# 延迟初始化GraphQL Schema
schema = None

def init_graphql_schema():
    global schema
    from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemySchema

    class UserType(SQLAlchemyObjectType):
        class Meta:
            model = get_user_model()
            exclude_fields = ('password',)

    class LinkType(SQLAlchemyObjectType):
        class Meta:
            model = get_link_model()

    # 定义Query
    class Query(graphene.ObjectType):
        all_users = graphene.List(UserType)
        all_links = graphene.List(LinkType)

        def resolve_all_users(self, info):
            return db.session.query(get_user_model()).all()

        def resolve_all_links(self, info):
            return db.session.query(get_link_model()).all()

    schema = SQLAlchemySchema(query=Query)

修改后的 app.py

from flask_graphql import GraphQLView
from database import init_db, schema
...
init_db(app)

# 现在schema已经初始化完成，可以正常使用
app.add_url_rule(
    "/graphql",
    view_func=GraphQLView.as_view(
    "graphql",
        schema=schema,
        graphiql=True,
    )
)

关键说明 #

• 把Schema的定义放进init_graphql_schema()函数里，只有在Base.prepare()执行完之后才调用这个函数，确保模型已经被反射出来。
• 避免在模块级别直接调用get_user_model()这类函数，因为模块导入时Base还没准备好。
• 如果原来的Schema是单独放在一个文件里的，同样要改成延迟加载的方式，比如用函数返回Schema，而不是直接导出一个提前创建好的对象。

内容的提问来源于stack exchange，提问作者Raylo