最新消息:雨落星辰是一个专注网站SEO优化、网站SEO诊断、搜索引擎研究、网络营销推广、网站策划运营及站长类的自媒体原创博客

Implement Pagination In Graphql instead of Graphene Python - Stack Overflow

programmeradmin0浏览0评论

here is my Gateway.py for Gateway Model

from datetime import datetime
import graphene
from models.v0100.util import get_query_types, stop_timestamps_w_timezone
from sqlalchemy.orm import joinedload
from graphene_sqlalchemy import SQLAlchemyObjectType
from graphene_sqlalchemy_filter import FilterSet

from . import SCHEMA_VERSION
from .core import DataDevice
from .orm import BuildingModel, ElectricMeterModel, GatewayModel, ElectricMeterPanelModel
from .util import apply_dynamic_eager_loading, get_nested_models, extract_pagination_args


class Gateway(SQLAlchemyObjectType):
    class Meta:
        model = GatewayModel
        interfaces = (graphene.relay.Node, DataDevice)
        exclude_fields = (
              '_building', '_building_id',
              '_electric_meter_panel',  '_electric_meter_panel_id',
              '_electric_meters', '_giai', '_name', '_native_id', '_serial_num',
              '_label', '_short_name', '_kit', '_singufm_asset_id', 'operational_status', '_id'

        )

    schemaVersion = graphene.String(description="The version of the schema for this object")
    def resolve_schemaVersion(self, info) -> graphene.String:
        return SCHEMA_VERSION

    building = graphene.Field(f'models.{SCHEMA_VERSION}.building.Building', required=True, description="The building that the Gateway is connected to")
    def resolve_building(self, info):
        if self._building_id is None:
            return None
        if self._building:
            return self._building

    electricMeterPanel = graphene.Field(f'models.{SCHEMA_VERSION}.electric_meter_panel.ElectricMeterPanel', required=False, description="The electric meter panel that the Gateway is connected to")
    def resolve_electricMeterPanel(self, info):
        if self._electric_meter_panel_id is None:
            return None
        if self._electric_meter_panel:
            return self._electric_meter_panel

    electricMeters = graphene.List(f'models.{SCHEMA_VERSION}.electric_meter.ElectricMeter', required=False, description="The Electric Meters that the Gateway is connected to")

    def resolve_electricMeters(self, info):
        return self._electric_meters or []

    @classmethod
    def query_resolver(cls, info, **kwargs):
        pasof = kwargs.pop('asof', datetime.utcnow())
        stop_timestamps_w_timezone(pasof)

        if not 'asof' in info.variable_values:
            info.variable_values['asof'] = pasof

        filters = kwargs.pop('filters', None)

        query = info.context['session'].query(GatewayModel).params({"p_asof": pasof})
        if filters:
            query = cls.Filter.filter(info, query, filters=filters)

        # type_names = get_query_types(info)
        print(f"Generated Query: {query}")

        # Apply eager loading for nested models
        models_names = get_nested_models(info)
        query = apply_dynamic_eager_loading(query, GatewayModel, models_names)
        data = extract_pagination_args(kwargs,query)
        print(f"Fetched data from query: {data}")
        return data
        
    class Filter(FilterSet):
        class Meta:
            name = 'GatewayFilter'
            model = GatewayModel
            fields = {
                'giai': [FilterSet.EQ]
                , 'nativeId': [FilterSet.EQ, FilterSet.ILIKE]
                , 'kit': [FilterSet.EQ, FilterSet.ILIKE]
                , 'serialNum': [FilterSet.EQ, FilterSet.ILIKE]
                , 'isOperational': [FilterSet.EQ]
                , 'operationalStatus': [FilterSet.EQ]

            }

and here is my extract_pagination_args :

    def decode_cursor(cursor):
    """Decode the cursor value from base64 and extract the offset."""
    if cursor and cursor.lower() != "none":
        decoded = base64.b64decode(cursor).decode("utf-8")
        return int(decoded.split(":")[-1])
    return None


def extract_pagination_args(kwargs, query):
    first = kwargs.pop('first', None)
    last = kwargs.pop('last', None)
    after_cursor = decode_cursor(kwargs.pop('after', None))
    end_cursor = decode_cursor(kwargs.pop('before', None))
    print("offset",after_cursor)

    total_count = query.count()
    if first:
        first =first+1
    if last:
        last = last+1
    if after_cursor:
        query = query.offset(after_cursor)
        if first:
            query = query.limit(first)
        else:
            query = query.limit(total_count - after_cursor)

    elif end_cursor:

        reverse_offset = max(0, total_count - after_cursor)
        query = query.offset(reverse_offset)
        if last:
            query = query.limit(last)
        else:
            query = query.limit(total_count - reverse_offset)

    elif first :
        # Forward pagination without cursor
        query = query.limit(first)

    elif last:
        # Fetch last N items without cursor
        start_offset = max(0, total_count - last)
        query = query.offset(start_offset).limit(last)

    return query.all()

while querying the first 10 am getting proper value but using that end cursor if i query the first:10 , after:"Cursor" am getting empty as graphene pagination again filter the same how to over come this issue

Any help would be appriciated

发布评论

评论列表(0)

  1. 暂无评论