如何从Django QuerySet动态填充GraphQL ENUM?
问题描述
原本我的GraphQL API中,Revenue字段使用常量定义的Currency枚举,代码如下:
from graphene import Enum CurrencyEnum = Enum('Currency', CURRENCY_CHOICES) class RevenueMetrics: revenue = graphene.Float(description='Revenue', currency=CurrencyEnum(required=True))
如今CURRENCY_CHOICES常量已迁移至数据库的Currency表中,需要从该表的QuerySet动态创建CurrencyEnum类以获取最新的货币集合,同时解决Schema重构相关问题。使用版本:
graphene==2.1.8 graphene-django==2.13.0
我尝试了以下代码,但仍需手动重新生成Schema:
def dcm(): return Enum('Currency', tuple((c, c) for c in Currency.objects.values_list('code', flat=True))) revenue = graphene.Float(description='Revenue', currency=dcm()(required=True))
解决方案
1. 动态生成枚举+Schema自动重载
Graphene的Schema默认在服务启动时生成,不会自动感知数据库变化,需从两方面处理:
动态枚举生成
将枚举生成逻辑封装为函数,避免在字段定义阶段直接执行数据库查询(防止启动时数据库未就绪或数据过时):
from graphene import Enum from django.apps import apps def get_currency_enum(): # 用apps.get_model避免循环导入问题 Currency = apps.get_model('你的应用名', 'Currency') # 从数据库拉取最新货币代码生成枚举选项 currency_options = [(item.code, item.code) for item in Currency.objects.all()] return Enum('Currency', currency_options) # 在字段定义中使用动态生成的枚举 class RevenueMetrics: revenue = graphene.Float(description='Revenue', currency=get_currency_enum()(required=True))
Schema自动重载(开发/生产区分处理)
- 开发环境:监听Currency表的增删改信号,触发Schema重新生成:
from django.db.models.signals import post_save, post_delete from django.dispatch import receiver from graphene_django.views import GraphQLView from 你的Schema模块 import generate_schema, schema @receiver([post_save, post_delete], sender=Currency) def refresh_graphql_schema(sender, instance, **kwargs): # 重新生成Schema global schema schema = generate_schema() # 更新GraphQL视图使用的Schema GraphQLView.schema = schema
注:需将Schema的生成逻辑封装到generate_schema()函数中,确保每次调用都会生成包含最新枚举的Schema。
- 生产环境:由于货币列表不会频繁变动,建议部署时预先生成枚举,或添加手动更新的管理命令:
# management/commands/refresh_currency_enum.py from django.core.management.base import BaseCommand from 你的Schema模块 import generate_schema class Command(BaseCommand): help = '刷新GraphQL Schema中的货币枚举' def handle(self, *args, **options): generate_schema() self.stdout.write(self.style.SUCCESS('货币枚举已成功刷新'))
2. 替代方案:自定义标量(高动态场景)
如果货币列表需要实时验证且不想频繁重构Schema,可改用自定义标量替代枚举,在请求阶段验证货币代码的有效性:
from graphene import Scalar from graphql.language.ast import StringValue from django.apps import apps class CurrencyScalar(Scalar): @staticmethod def serialize(value): return str(value) @staticmethod def parse_literal(ast): if isinstance(ast, StringValue): Currency = apps.get_model('你的应用名', 'Currency') if Currency.objects.filter(code=ast.value).exists(): return ast.value raise ValueError(f"无效的货币代码: {ast.value}") @staticmethod def parse_value(value): Currency = apps.get_model('你的应用名', 'Currency') if Currency.objects.filter(code=value).exists(): return value raise ValueError(f"无效的货币代码: {value}") # 在字段中使用自定义标量 class RevenueMetrics: revenue = graphene.Float(description='Revenue', currency=CurrencyScalar(required=True))
该方案无需重构Schema,但会失去GraphQL Playground中的枚举自动提示功能。
内容的提问来源于stack exchange,提问作者Danialgo
相关产品推荐
相关产品推荐

