Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/_base/mixins.py: 87%

1from collections.abc import Callable 1 ctx1t

2from logging import Logger 1 ctx1t

3 

4import sqlalchemy.exc 1 ctx1t

5from fastapi import HTTPException, status 1 ctx1t

6from pydantic import UUID4, BaseModel 1 ctx1t

7 

8from mealie.repos.repository_generic import RepositoryGeneric 1 ctx1t

9from mealie.schema.response import ErrorResponse 1 ctx1t

10 

11 

12class HttpRepo[C: BaseModel, R: BaseModel, U: BaseModel]: 1 ctx1t

13 """ 

14 The HttpRepo[C, R, U] class is a mixin class that provides a common set of methods for CRUD operations. 

15 This class is intended to be used in a composition pattern where a class has a mixin property. For example: 

16 

17 ``` 

18 class MyClass: 

19 def __init__(self, repo, logger): 

20 self.mixins = HttpRepo(repo, logger) 

21 ``` 

22 

23 """ 

24 

25 repo: RepositoryGeneric 1 ctx1t

26 exception_msgs: Callable[[type[Exception]], str] | None 1 ctx1t

27 default_message: str = "An unexpected error occurred." 1 ctx1t

28 

29 def __init__( 1 ctx1t

30 self, 

31 repo: RepositoryGeneric, 

32 logger: Logger, 

33 exception_msgs: Callable[[type[Exception]], str] | None = None, 

34 default_message: str | None = None, 

35 ) -> None: 

36 self.repo = repo 45 ctx1CKDIERLTSfpqulvbhmwMiyNgnzodcAjerJOFGPHksBQxa

37 self.logger = logger 45 ctx1CKDIERLTSfpqulvbhmwMiyNgnzodcAjerJOFGPHksBQxa

38 self.exception_msgs = exception_msgs 45 ctx1CKDIERLTSfpqulvbhmwMiyNgnzodcAjerJOFGPHksBQxa

39 

40 if default_message: 45 ctx1CKDIERLTSfpqulvbhmwMiyNgnzodcAjerJOFGPHksBQxa

41 self.default_message = default_message 22 ctx1fpqulvbhmwignzdcAekBxa

42 

43 def get_exception_message(self, ext: Exception) -> str: 1 ctx1t

44 if self.exception_msgs: 34 ctx1CDIEfpqulvbhmwiygnzodcAjerFGHksBxa

45 return self.exception_msgs(type(ext)) 33 ctx1CIEfpqulvbhmwiygnzodcAjerFGHksBxa

46 return self.default_message 18 ctx1Dfulvbhmwiynoderxa

47 

48 def handle_exception(self, ex: Exception) -> None: 1 ctx1t

49 # Cleanup 

50 self.logger.exception(ex) 34 ctx1CDIEfpqulvbhmwiygnzodcAjerFGHksBxa

51 self.repo.session.rollback() 34 ctx1CDIEfpqulvbhmwiygnzodcAjerFGHksBxa

52 

53 # Respond 

54 msg = self.get_exception_message(ex) 34 ctx1CDIEfpqulvbhmwiygnzodcAjerFGHksBxa

55 

56 if isinstance(ex, sqlalchemy.exc.NoResultFound): 34 ctx1CDIEfpqulvbhmwiygnzodcAjerFGHksBxa

57 raise HTTPException( 5 ctx1Ebdea

58 status.HTTP_404_NOT_FOUND, 

59 detail=ErrorResponse.respond(message=msg, exception=str(ex)), 

60 ) 

61 else: 

62 raise HTTPException( 34 ctx1CDIEfpqulvbhmwiygnzodcAjerFGHksBxa

63 status.HTTP_400_BAD_REQUEST, 

64 detail=ErrorResponse.respond(message=msg, exception=str(ex)), 

65 ) 

66 

67 def create_one(self, data: C) -> R | None: 1 ctx1t

68 item: R | None = None 40 ctx1CKDLfpqulvbhmwMiyNgnzodcAjerJOFGPHksBQxa

69 try: 40 ctx1CKDLfpqulvbhmwMiyNgnzodcAjerJOFGPHksBQxa

70 item = self.repo.create(data) 40 ctx1CKDLfpqulvbhmwMiyNgnzodcAjerJOFGPHksBQxa

71 except Exception as ex: 32 ctx1CDfpqulvbhmwiygnzodcAjerFGHksBxa

72 self.handle_exception(ex) 32 ctx1CDfpqulvbhmwiygnzodcAjerFGHksBxa

73 

74 return item 40 ctx1CKDLfpqulvbhmwMiyNgnzodcAjerJOFGPHksBQxa

75 

76 def get_one(self, item_id: int | str | UUID4, key: str | None = None) -> R: 1 ctx1t

77 item = self.repo.get_one(item_id, key) 12 ctx1Rbhgdcjerksa

78 

79 if not item: 12 ctx1Rbhgdcjerksa

80 raise HTTPException( 6 ctx1Rbcjka

81 status.HTTP_404_NOT_FOUND, 

82 detail=ErrorResponse.respond(message="Not found."), 

83 ) 

84 

85 return item 8 ctx1hgdjersa

86 

87 def update_one(self, data: U, item_id: int | str | UUID4) -> R: 1 ctx1t

88 item = self.repo.get_one(item_id) 12 ctx1TSfplbhmgoca

89 

90 if not item: 11 ctx1Sfplbhmgoca

91 raise HTTPException( 2 ctx1Sa

92 status.HTTP_404_NOT_FOUND, 

93 detail=ErrorResponse.respond(message="Not found."), 

94 ) 

95 

96 try: 10 ctx1fplbhmgoca

97 item = self.repo.update(item_id, data) # type: ignore 10 ctx1fplbhmgoca

98 except Exception as ex: 3 ctx1goa

99 self.handle_exception(ex) 3 ctx1goa

100 

101 return item 9 ctx1fplbhmgca

102 

103 def patch_one(self, data: U, item_id: int | str | UUID4) -> R: 1 ctx1t

104 item = self.repo.get_one(item_id) 

105 

106 if not item: 

107 raise HTTPException( 

108 status.HTTP_404_NOT_FOUND, 

109 detail=ErrorResponse.respond(message="Not found."), 

110 ) 

111 

112 try: 

113 item = self.repo.patch(item_id, data.model_dump(exclude_unset=True, exclude_defaults=True)) 

114 except Exception as ex: 

115 self.handle_exception(ex) 

116 

117 return item 

118 

119 def delete_one(self, item_id: int | str | UUID4) -> R | None: 1 ctx1t

120 item: R | None = None 14 ctx1IEfqbindcjeJka

121 try: 14 ctx1IEfqbindcjeJka

122 item = self.repo.delete(item_id) 14 ctx1IEfqbindcjeJka

123 self.logger.info(f"Deleting item with id {item_id}") 12 ctx1fqbindcjeJka

124 except Exception as ex: 7 ctx1IEbidea

125 self.handle_exception(ex) 7 ctx1IEbidea

126 

127 return item 12 ctx1fqbindcjeJka