Coverage for api/errors.py: 26%

1import logging 

2import traceback 

3from functools import wraps 

4 

5from config import LOG_LEVEL 

6from fastapi import HTTPException, Request 

7from fastapi.responses import JSONResponse 

8from jose.exceptions import ExpiredSignatureError 

9from logging_config import logger 

10from pydantic_core import ValidationError 

11from services.telegram import group_notification 

12from sotrans_fastapi_keycloak import KeycloakError 

13 

14 

15def log_error(add_text=""): 

16 if LOG_LEVEL > logging.ERROR: 

17 return 

18 err = traceback.format_exc() 

19 if add_text: 

20 err += f"\n\n{add_text}" 

21 logger.error(err) 

22 group_notification(err, logging.ERROR) 

23 

24 

25def log_warning(add_text=""): 

26 if LOG_LEVEL > logging.WARNING: 

27 return 

28 err = traceback.format_exc() 

29 if err == "NoneType: None\n": 

30 err = "" 

31 if add_text: 

32 err += f"\n\n{add_text}" 

33 logger.warning(err) 

34 group_notification(err, logging.WARNING) 

35 

36 

37def error_handler(func): 

38 @wraps(func) 

39 async def wrapped(*args, **kwargs): 

40 try: 

41 result = await func(*args, **kwargs) 

42 return result 

43 except Exception: 

44 log_error() 

45 

46 return wrapped 

47 

48 

49def endpoint_error_handler(func): 

50 @wraps(func) 

51 async def wrapped(*args, **kwargs): 

52 try: 

53 result = await func(*args, **kwargs) 

54 return result 

55 except HTTPException as e: 

56 raise e 

57 except ValidationError as e: 

58 raise HTTPException(status_code=422, detail=str(e)) 

59 except KeycloakError as e: 

60 keycloak_error_handler(None, e) 

61 except ExpiredSignatureError as e: 

62 logger.info("HERE") 

63 raise HTTPException(status_code=401, detail=str(e)) 

64 except Exception: 

65 log_error() 

66 raise HTTPException( 

67 status_code=500, detail={"error": traceback.format_exc()} 

68 ) 

69 

70 return wrapped 

71 

72 

73def keycloak_error_handler(request: Request | None, exc: KeycloakError): 

74 return JSONResponse( 

75 status_code=exc.status_code, 

76 content={"сообщение": exc.reason}, 

77 ) 

78 

79 

80def app_error_handler(request: Request | None, exc: Exception): 

81 match exc: 

82 case ExpiredSignatureError(): # type: ignore[misc] 

83 return JSONResponse( 

84 status_code=401, 

85 content={"message": str(exc)}, 

86 ) 

87 case _: 

88 return JSONResponse( 

89 status_code=500, 

90 content={"message": str(exc)}, 

91 )