94 lines
2.7 KiB
Python
94 lines
2.7 KiB
Python
from typing import Any, Callable, List, Optional, Tuple, Type
|
|
|
|
from django.conf import settings
|
|
from django.db.models import QuerySet
|
|
from loguru import logger
|
|
from ninja import NinjaAPI, Schema
|
|
from ninja.pagination import PageNumberPagination as NinjaPageNumberPagination
|
|
from ninja.security import HttpBearer
|
|
|
|
from takahe.utils import Takahe
|
|
from users.models.apidentity import APIdentity
|
|
|
|
PERMITTED_WRITE_METHODS = ["PUT", "POST", "DELETE", "PATCH"]
|
|
PERMITTED_READ_METHODS = ["GET", "HEAD", "OPTIONS"]
|
|
|
|
|
|
class OAuthAccessTokenAuth(HttpBearer):
|
|
def authenticate(self, request, token) -> bool:
|
|
if not token:
|
|
logger.debug("API auth: no access token provided")
|
|
return False
|
|
tk = Takahe.get_token(token)
|
|
if not tk:
|
|
logger.debug("API auth: access token not found")
|
|
return False
|
|
request_scope = ""
|
|
request_method = request.method
|
|
if request_method in PERMITTED_READ_METHODS:
|
|
request_scope = "read"
|
|
elif request_method in PERMITTED_WRITE_METHODS:
|
|
request_scope = "write"
|
|
else:
|
|
logger.debug("API auth: unsupported HTTP method")
|
|
return False
|
|
if request_scope not in tk.scopes:
|
|
logger.debug("API auth: scope not allowed")
|
|
return False
|
|
identity = APIdentity.objects.filter(pk=tk.identity_id).first()
|
|
if not identity:
|
|
logger.debug("API auth: identity not found")
|
|
return False
|
|
if identity.deleted:
|
|
logger.debug("API auth: identity deleted")
|
|
return False
|
|
user = identity.user
|
|
if not user:
|
|
logger.debug("API auth: user not found")
|
|
return False
|
|
request.user = user
|
|
return True
|
|
|
|
|
|
class EmptyResult(Schema):
|
|
pass
|
|
|
|
|
|
class Result(Schema):
|
|
message: str | None
|
|
# error: Optional[str]
|
|
|
|
|
|
class RedirectedResult(Schema):
|
|
message: str | None
|
|
url: str
|
|
|
|
|
|
class PageNumberPagination(NinjaPageNumberPagination):
|
|
items_attribute = "data"
|
|
|
|
class Output(Schema):
|
|
data: List[Any]
|
|
pages: int
|
|
count: int
|
|
|
|
def paginate_queryset(
|
|
self,
|
|
queryset: QuerySet,
|
|
pagination: NinjaPageNumberPagination.Input,
|
|
**params: Any,
|
|
):
|
|
val = super().paginate_queryset(queryset, pagination, **params)
|
|
return {
|
|
"data": val["items"],
|
|
"count": val["count"],
|
|
"pages": (val["count"] + self.page_size - 1) // self.page_size,
|
|
}
|
|
|
|
|
|
api = NinjaAPI(
|
|
auth=OAuthAccessTokenAuth(),
|
|
title=f'{settings.SITE_INFO["site_name"]} API',
|
|
version="1.0.0",
|
|
description=f"{settings.SITE_INFO['site_name']} API <hr/><a href='{settings.SITE_INFO['site_url']}'>Learn more</a>",
|
|
)
|