mirror of https://gitlab.com/tildes/tildes.git
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
86 lines
2.7 KiB
86 lines
2.7 KiB
# Copyright (c) 2018 Tildes contributors <code@tildes.net>
|
|
# SPDX-License-Identifier: AGPL-3.0-or-later
|
|
|
|
"""JSON API utils."""
|
|
|
|
from typing import Tuple
|
|
from pyramid.request import Request
|
|
from pyramid.response import Response
|
|
from tildes.lib.id import split_anchored_id
|
|
from tildes.models.pagination import PaginatedQuery, PaginatedResults
|
|
|
|
|
|
def query_apply_pagination( # noqa
|
|
query, before, after, error_if_no_anchor: bool = False
|
|
) -> PaginatedQuery:
|
|
"""Apply pagination parameters to a query."""
|
|
# Start by parsing the before/after parameters and extracting the anchor type
|
|
# We don't know if the ID has an anchor, so we just try to split it
|
|
# If it doesn't have an anchor, we just use the ID as is.
|
|
anchor_type = None
|
|
if before and after:
|
|
raise ValueError("Cannot specify both before and after parameters")
|
|
if before:
|
|
try:
|
|
anchor_type, before = split_anchored_id(before)
|
|
except ValueError as exc:
|
|
if error_if_no_anchor:
|
|
raise ValueError(
|
|
"Expected an anchored ID for 'before' parameter"
|
|
) from exc
|
|
if after:
|
|
try:
|
|
anchor_type, after = split_anchored_id(after)
|
|
except ValueError as exc:
|
|
if error_if_no_anchor:
|
|
raise ValueError(
|
|
"Expected an anchored ID for 'after' parameter"
|
|
) from exc
|
|
|
|
if anchor_type:
|
|
query = query.anchor_type(anchor_type)
|
|
if before:
|
|
query = query.before_id36(before)
|
|
if after:
|
|
query = query.after_id36(after)
|
|
return query
|
|
|
|
|
|
def get_next_and_prev_link(request: Request, page: PaginatedResults) -> Tuple[str, str]:
|
|
"""Get the next and previous links for pagination."""
|
|
next_link = None
|
|
prev_link = None
|
|
|
|
if page.has_next_page:
|
|
query_vars = request.GET.copy()
|
|
query_vars.pop("before", None)
|
|
query_vars.update({"after": page.next_page_after_id36})
|
|
next_link = request.current_route_url(_query=query_vars)
|
|
|
|
if page.has_prev_page:
|
|
query_vars = request.GET.copy()
|
|
query_vars.pop("after", None)
|
|
query_vars.update({"before": page.prev_page_before_id36})
|
|
prev_link = request.current_route_url(_query=query_vars)
|
|
|
|
return (next_link, prev_link)
|
|
|
|
|
|
def build_error_response(
|
|
message: str,
|
|
status: int = 400,
|
|
field: str = "N/A",
|
|
error_type: str = "ValidationError",
|
|
) -> Response:
|
|
"""Build a standardized error response."""
|
|
return Response(
|
|
status=status,
|
|
content_type="application/json",
|
|
json=[
|
|
{
|
|
"message": message,
|
|
"field": field,
|
|
"exception": error_type,
|
|
}
|
|
],
|
|
)
|