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