Coverage for src\utility\lib.py: 100%

1from fastapi import HTTPException, status 

2from fastapi.responses import JSONResponse 

3from typing import Any 

4import logging 

5import msgspec 

6 

7 

8class CustomException: 

9 @staticmethod 

10 def not_found(detail: str = "The requested resource could not be found.") -> HTTPException: 

11 return HTTPException( 

12 status_code=status.HTTP_404_NOT_FOUND, 

13 detail=detail, 

14 ) 

15 

16 @staticmethod 

17 def server_error(detail: str = "The server encountered an unexpected error.") -> HTTPException: 

18 return HTTPException( 

19 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, 

20 detail=detail, 

21 ) 

22 

23 @staticmethod 

24 def service_unavailable(detail: str = "The service is currently unavailable.") -> HTTPException: 

25 return HTTPException( 

26 status_code=status.HTTP_503_SERVICE_UNAVAILABLE, 

27 detail=detail, 

28 ) 

29 

30 

31class Logger: 

32 formatter = logging.Formatter( 

33 fmt="%(asctime)s [%(levelname)s] %(message)s", 

34 datefmt="%Y-%m-%d %H:%M:%S" 

35 ) 

36 

37 handler = logging.StreamHandler() 

38 handler.setFormatter(formatter) 

39 handler.setLevel(logging.INFO) 

40 

41 logger = logging.getLogger(__name__) 

42 logger.addHandler(handler) 

43 logger.setLevel(logging.INFO) 

44 

45 @classmethod 

46 def info(cls, message: str) -> None: 

47 cls.logger.info(message) 

48 

49 @classmethod 

50 def warn(cls, message: str) -> None: 

51 cls.logger.warning(message) 

52 

53 @classmethod 

54 def err(cls, message: str) -> None: 

55 cls.logger.error(message) 

56 

57 @classmethod 

58 def exception(cls, message: str) -> None: 

59 cls.logger.exception(message) 

60 

61 

62class MsgSpecJSONResponse(JSONResponse): 

63 @staticmethod 

64 def render(content: Any) -> bytes: 

65 return msgspec.json.encode(content) 

66 

67 @staticmethod 

68 def parse(content: Any, type: Any) -> Any: 

69 return msgspec.json.decode(content, type=type)