etesync-server/etebase_fastapi/utils.py

86 lines
2.2 KiB
Python
Raw Normal View History

2020-12-27 16:54:06 +00:00
import dataclasses
import typing as t
from typing_extensions import Literal
2020-12-28 11:56:53 +00:00
import msgpack
2020-12-28 12:26:44 +00:00
import base64
2020-12-27 16:54:06 +00:00
from fastapi import status, Query, Depends
from pydantic import BaseModel as PyBaseModel
2020-12-29 15:18:09 +00:00
from django.db.models import Model, QuerySet
from django.core.exceptions import ObjectDoesNotExist
from django_etebase import app_settings
from django_etebase.models import AccessLevels
2020-12-29 11:22:36 +00:00
from myauth.models import UserType, get_typed_user_model
2020-12-28 08:00:35 +00:00
from .exceptions import HttpError, HttpErrorOut
2020-12-29 11:22:36 +00:00
User = get_typed_user_model()
2020-12-27 16:54:06 +00:00
Prefetch = Literal["auto", "medium"]
2020-12-27 16:54:06 +00:00
PrefetchQuery = Query(default="auto")
2020-12-29 15:18:09 +00:00
T = t.TypeVar("T", bound=Model, covariant=True)
class BaseModel(PyBaseModel):
class Config:
json_encoders = {
bytes: lambda x: x,
}
2020-12-27 16:54:06 +00:00
@dataclasses.dataclass
class Context:
2020-12-29 11:22:36 +00:00
user: t.Optional[UserType]
2020-12-27 16:54:06 +00:00
prefetch: t.Optional[Prefetch]
2020-12-29 15:18:09 +00:00
def get_object_or_404(queryset: QuerySet[T], **kwargs) -> T:
try:
return queryset.get(**kwargs)
except ObjectDoesNotExist as e:
2020-12-28 07:51:27 +00:00
raise HttpError("does_not_exist", str(e), status_code=status.HTTP_404_NOT_FOUND)
def is_collection_admin(collection, user):
member = collection.members.filter(user=user).first()
return (member is not None) and (member.accessLevel == AccessLevels.ADMIN)
2020-12-28 08:00:35 +00:00
2021-01-04 09:56:17 +00:00
def msgpack_encode(content) -> bytes:
ret = msgpack.packb(content, use_bin_type=True)
assert ret is not None
return ret
2020-12-28 11:56:53 +00:00
2021-01-04 09:56:17 +00:00
def msgpack_decode(content: bytes):
2020-12-28 11:56:53 +00:00
return msgpack.unpackb(content, raw=False)
2021-01-04 09:56:17 +00:00
def b64encode(value: bytes):
2020-12-28 12:26:44 +00:00
return base64.urlsafe_b64encode(value).decode("ascii").strip("=")
2021-01-04 09:56:17 +00:00
def b64decode(data: str):
2020-12-28 12:26:44 +00:00
data += "=" * ((4 - len(data) % 4) % 4)
return base64.urlsafe_b64decode(data)
def get_user_username_email_kwargs(username: str):
field_name = User.EMAIL_FIELD if "@" in username else User.USERNAME_FIELD
return {field_name + "__iexact": username.lower()}
PERMISSIONS_READ = [Depends(x) for x in app_settings.API_PERMISSIONS_READ]
PERMISSIONS_READWRITE = PERMISSIONS_READ + [Depends(x) for x in app_settings.API_PERMISSIONS_WRITE]
2020-12-28 08:00:35 +00:00
response_model_dict = {"model": HttpErrorOut}
2020-12-28 13:26:34 +00:00
permission_responses: t.Dict[t.Union[int, str], t.Dict[str, t.Any]] = {
401: response_model_dict,
403: response_model_dict,
}