GraphQL Django：含自定义字段的Type如何在解析器中缓存

问题 #

使用graphene-django结合Redis缓存时遇到问题：缓存的AdDetails查询集不包含自定义的价格相关字段，因为缓存的是模型实例而非对应的AdDetailsType对象，再次查询时仍需重新计算自定义字段，耗时较长。尝试[AdDetailsType(ad) for ad in result]转换后所有字段值为None，序列化方法也未解决。

相关代码 #

模型定义

class AdDetails(models.Model):
    id = models.IntegerField(primary_key=True)
    dealer = models.ForeignKey('DealerLookup', models.DO_NOTHING, db_column='dealer')
    county = models.ForeignKey('CountyLookup', models.DO_NOTHING, db_column='county')
    vehicle_type = models.ForeignKey('VehicleTypeLookup', models.DO_NOTHING, db_column='vehicle_type')
    href = models.TextField()
    make = models.ForeignKey('MakeLookup', models.DO_NOTHING, db_column='make')
    model = models.ForeignKey('ModelLookup', models.DO_NOTHING, db_column='model')
    year = models.ForeignKey('YearLookup', models.DO_NOTHING, db_column='year')
    mileage_km = models.IntegerField(blank=True, null=True)
    fuel_type = models.ForeignKey('FuelTypeLookup', models.DO_NOTHING, db_column='fuel_type')
    transmission = models.ForeignKey('TransmissionLookup', models.DO_NOTHING, db_column='transmission', blank=True, null=True)
    engine_size = models.ForeignKey('EngineSizeLookup', models.DO_NOTHING, db_column='engine_size', blank=True, null=True)
    colour = models.ForeignKey('ColourLookup', models.DO_NOTHING, db_column='colour', blank=True, null=True)
    for_sale_date = models.DateField()
    sold_date = models.DateField(blank=True, null=True)

    class Meta:
        managed = False
        db_table = 'ad_details'

AdDetailsType定义

class AdDetailsType(DjangoObjectType):
    price = graphene.Float()
    previous_price = graphene.Float()
    price_movement = graphene.Float()

    def resolve_price(self, info):
        price_history = self.pricehistory_set.order_by('-date').first()
        if price_history:
            return price_history.price
        return None

    def resolve_previous_price(self, info):
        price_history = self.pricehistory_set.filter(price__gt=0).order_by('date').first()
        if price_history:
            return price_history.price
        return None

    def resolve_price_movement(self, info):
        price = self.pricehistory_set.order_by('-date').first()
        prev_price = self.pricehistory_set.filter(price__gt=0).order_by('date').first()

        if price is not None and prev_price is not None:
            return price.price - prev_price.price
        else:
            return None

    class Meta:
        model = AdDetails
        fields = "__all__"

当前解析器代码

class Query(graphene.ObjectType):
    stock_details = graphene.List(AdDetailsType, dealer_id=graphene.Int())

    def resolve_stock_details(self, info, dealer_id):
        cache_key = f"stock_details_{dealer_id}"
        cached_result = cache.get(cache_key)
        if cached_result is not None:
            return cached_result

        result = AdDetails.objects.filter(dealer=dealer_id, sold_date__isnull=True).order_by('for_sale_date')
        
        # Cache the query result with a timeout of 10 days
        cache.set(cache_key, result, timeout=864000)

        return result

解决方案 #

方案一：缓存包含自定义字段的序列化数据 #

直接计算好所有自定义字段的值，把完整数据序列化为JSON存入缓存，取出后直接构造AdDetailsType返回，无需再执行resolve方法。

修改后的解析器代码：

import json
from django.core.serializers.json import DjangoJSONEncoder

class Query(graphene.ObjectType):
    stock_details = graphene.List(AdDetailsType, dealer_id=graphene.Int())

    def resolve_stock_details(self, info, dealer_id):
        cache_key = f"stock_details_{dealer_id}"
        cached_result = cache.get(cache_key)
        if cached_result is not None:
            # 反序列化缓存数据并构造AdDetailsType实例
            return [AdDetailsType(**item) for item in json.loads(cached_result)]

        # 预取pricehistory集合，避免N+1查询
        result = AdDetails.objects.filter(dealer=dealer_id, sold_date__isnull=True)\
                                  .order_by('for_sale_date')\
                                  .prefetch_related('pricehistory_set')

        # 构造包含模型字段和自定义字段的字典列表
        serialized_data = []
        for ad in result:
            # 提取模型字段数据
            ad_data = {field.name: getattr(ad, field.name) for field in AdDetails._meta.fields}
            # 计算自定义字段值
            latest_price = ad.pricehistory_set.order_by('-date').first()
            first_valid_price = ad.pricehistory_set.filter(price__gt=0).order_by('date').first()
            
            ad_data['price'] = latest_price.price if latest_price else None
            ad_data['previous_price'] = first_valid_price.price if first_valid_price else None
            ad_data['price_movement'] = (latest_price.price - first_valid_price.price) if (latest_price and first_valid_price) else None
            
            serialized_data.append(ad_data)
        
        # 序列化并存入Redis缓存
        cache.set(cache_key, json.dumps(serialized_data, cls=DjangoJSONEncoder), timeout=864000)

        return [AdDetailsType(**item) for item in serialized_data]

方案二：预取关联数据，缓存模型实例（优化DB查询） #

如果不想直接缓存序列化后的Type对象，可以预取pricehistory_set关联数据，减少resolve阶段的数据库查询次数。即使缓存的是模型实例，后续resolve时也无需再次查询数据库，大幅提升性能。

修改后的解析器代码：

class Query(graphene.ObjectType):
    stock_details = graphene.List(AdDetailsType, dealer_id=graphene.Int())

    def resolve_stock_details(self, info, dealer_id):
        cache_key = f"stock_details_{dealer_id}"
        cached_result = cache.get(cache_key)
        if cached_result is not None:
            return cached_result

        # 预取pricehistory集合，避免resolve时的N+1查询
        result = AdDetails.objects.filter(dealer=dealer_id, sold_date__isnull=True)\
                                  .order_by('for_sale_date')\
                                  .prefetch_related('pricehistory_set')
        
        # 缓存模型实例查询集
        cache.set(cache_key, result, timeout=864000)

        return result

关于直接转换AdDetailsType失败的原因 #

AdDetailsType(ad)无法正确赋值是因为DjangoObjectType的初始化需要遵循graphene内部的字段绑定机制，直接传入模型实例不会自动映射所有字段。正确的做法要么是通过序列化后的字典数据构造，要么让graphene自动处理预取了关联数据的模型实例。

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