etesync-server/etebase_fastapi/stoken_handler.py

63 lines
1.8 KiB
Python
Raw Normal View History

2020-12-25 09:52:43 +00:00
import typing as t
from django.db.models import QuerySet
from fastapi import status
from django_etebase.models import Stoken
from .exceptions import HttpError
2020-12-25 09:52:43 +00:00
# TODO missing stoken_annotation type
StokenAnnotation = t.Any
2020-12-27 13:25:16 +00:00
def get_stoken_obj(stoken: t.Optional[str]) -> t.Optional[Stoken]:
if stoken:
2020-12-25 09:52:43 +00:00
try:
return Stoken.objects.get(uid=stoken)
except Stoken.DoesNotExist:
raise HttpError("bad_stoken", "Invalid stoken.", status_code=status.HTTP_400_BAD_REQUEST)
2020-12-25 09:52:43 +00:00
return None
def filter_by_stoken(
stoken: t.Optional[str], queryset: QuerySet, stoken_annotation: StokenAnnotation
2020-12-27 13:25:16 +00:00
) -> t.Tuple[QuerySet, t.Optional[Stoken]]:
2020-12-25 09:52:43 +00:00
stoken_rev = get_stoken_obj(stoken)
queryset = queryset.annotate(max_stoken=stoken_annotation).order_by("max_stoken")
if stoken_rev is not None:
queryset = queryset.filter(max_stoken__gt=stoken_rev.id)
return queryset, stoken_rev
2020-12-27 15:58:15 +00:00
def get_queryset_stoken(queryset: t.Iterable[t.Any]) -> t.Optional[Stoken]:
2020-12-25 09:52:43 +00:00
maxid = -1
for row in queryset:
rowmaxid = getattr(row, "max_stoken") or -1
maxid = max(maxid, rowmaxid)
2020-12-27 13:26:36 +00:00
new_stoken = Stoken.objects.get(id=maxid) if (maxid >= 0) else None
2020-12-25 09:52:43 +00:00
return new_stoken or None
def filter_by_stoken_and_limit(
stoken: t.Optional[str], limit: int, queryset: QuerySet, stoken_annotation: StokenAnnotation
) -> t.Tuple[list, t.Optional[Stoken], bool]:
queryset, stoken_rev = filter_by_stoken(stoken=stoken, queryset=queryset, stoken_annotation=stoken_annotation)
result = list(queryset[: limit + 1])
if len(result) < limit + 1:
done = True
else:
done = False
result = result[:-1]
new_stoken_obj = get_queryset_stoken(result) or stoken_rev
return result, new_stoken_obj, done