I have a pagination logic written in django which needs to convert in sqlalchemy. I am new to sqlalchemy. I read multiple articles but still could not figure out the following logic which is as follow
def _prepare_filter_expression(
field_name: str,
index: int,
cursor: List[str],
sorting_fields: List[str],
sorting_direction: str,
) -> Tuple[Q, Dict[str, Union[str, bool]]]:
field_expression: Dict[str, Union[str, bool]] = {}
extra_expression = Q()
for cursor_id, cursor_value in enumerate(cursor[:index]):
field_expression[sorting_fields[cursor_id]] = cursor_value
if sorting_direction == "gt":
extra_expression |= Q(**{f"{field_name}__{sorting_direction}": cursor[index]})
extra_expression |= Q(**{f"{field_name}__isnull": True})
elif cursor[index] is not None:
field_expression[f"{field_name}__{sorting_direction}"] = cursor[index]
else:
field_expression[f"{field_name}__isnull"] = False
return extra_expression, field_expression
def _prepare_filter(
cursor: List[str], sorting_fields: List[str], sorting_direction: str
) -> Q:
"""Create filter arguments based on sorting fields.
:param cursor: list of values that are passed from page_info, used for filtering.
:param sorting_fields: list of fields that were used for sorting.
:param sorting_direction: keyword direction ('lt', gt').
:return: Q() in following format
(OR: ('first_field__gt', 'first_value_form_cursor'),
(AND: ('second_field__gt', 'second_value_form_cursor'),
('first_field', 'first_value_form_cursor')),
(AND: ('third_field__gt', 'third_value_form_cursor'),
('second_field', 'second_value_form_cursor'),
('first_field', 'first_value_form_cursor'))
)
"""
filter_kwargs = Q()
for index, field_name in enumerate(sorting_fields):
if cursor[index] is None and sorting_direction == "gt":
continue
extra_expression, field_expression = _prepare_filter_expression(
field_name, index, cursor, sorting_fields, sorting_direction
)
filter_kwargs |= Q(extra_expression, **field_expression)
return filter_kwargs
When using django, what i am doing is I pass queryset along with arguments for sorting based on fields and direction so further query will be applied in the logic above.
So, my question is how can i write the following query in sqlalchemy
extra_expression = Q()
extra_expression |= Q(**{f"{field_name}__{sorting_direction}": cursor[index]})
Here is the gist for full code
https://gist.github.com/SanskarSans/a60b4c2244fd6217e5babf64e37fc849