73 lines
2.5 KiB
Python
73 lines
2.5 KiB
Python
from typing import Mapping, Optional, Self
|
|
from fastapi import HTTPException, Request, Response
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.utils import is_body_allowed_for_status_code
|
|
from fastapi.encoders import jsonable_encoder
|
|
from fastapi import status
|
|
from pydantic import HttpUrl
|
|
from starlette.datastructures import URL
|
|
|
|
from dooris_api import models
|
|
|
|
|
|
class HttpProblemException(HTTPException):
|
|
problem: models.HttpProblemDetail
|
|
headers: Optional[Mapping[str, str]]
|
|
|
|
def __init__(
|
|
self,
|
|
problem: models.HttpProblemDetail,
|
|
headers: Optional[Mapping[str, str]] = None,
|
|
):
|
|
self.problem = problem
|
|
super().__init__(status_code=problem.status, headers=headers)
|
|
|
|
def __str__(self) -> str:
|
|
return str(self.problem)
|
|
|
|
@classmethod
|
|
def unauthorized(cls, request_uri: str | URL) -> Self:
|
|
return cls(
|
|
problem=models.HttpProblemDetail(
|
|
type=models.HttpProblemType.UNAUTHORIZED,
|
|
status=status.HTTP_401_UNAUTHORIZED,
|
|
title="Unauthorized",
|
|
detail="You tried to access a ressource which requires authentication but you are not authenticated",
|
|
instance=HttpUrl(str(request_uri)),
|
|
)
|
|
)
|
|
|
|
@classmethod
|
|
def lock_not_found(cls, requested_lock: str, request_uri: str | URL) -> Self:
|
|
return cls(
|
|
problem=models.HttpProblemDetail(
|
|
type=models.HttpProblemType.LOCK_NOT_FOUND,
|
|
status=status.HTTP_404_NOT_FOUND,
|
|
title="Lock not found",
|
|
detail=f"You tried to interact with lock {requested_lock!r} that is not known to dooris",
|
|
instance=str(request_uri),
|
|
)
|
|
)
|
|
|
|
@classmethod
|
|
def forbidden_to_operate(cls, request_uri: str | URL) -> Self:
|
|
return cls(
|
|
problem=models.HttpProblemDetail(
|
|
type=models.HttpProblemType.FORBIDDEN_TO_OPERATE,
|
|
status=status.HTTP_403_FORBIDDEN,
|
|
title="Forbidden to operate locks",
|
|
detail="You are not allowed to operate locks",
|
|
instance=str(request_uri),
|
|
)
|
|
)
|
|
|
|
|
|
async def problem_exception_handler(
|
|
request: Request, exc: HttpProblemException
|
|
) -> Response:
|
|
headers = exc.headers
|
|
if not is_body_allowed_for_status_code(exc.problem.status):
|
|
return Response(status_code=exc.status_code, headers=headers)
|
|
return JSONResponse(
|
|
jsonable_encoder(exc.problem), status_code=exc.status_code, headers=exc.headers
|
|
)
|