Coverage for src/app/repositories/support_message_postgres_repository.py: 100%

1import psycopg2 

2from entities.support_message_entity import SupportMessageEntity 

3 

4class SupportMessagePostgresRepository: 

5 def __init__(self, db_config: dict): 

6 ''' 

7 Initializes the repository with the given database configuration. 

8 Args: 

9 db_config (dict): A dictionary containing the PostgreSQL database configuration parameters. 

10 ''' 

11 self.__db_config = db_config 

12 

13 def __connect(self): 

14 ''' 

15 Creates and returns a new connection to the PostgreSQL database. 

16 Returns: 

17 psycopg2.extensions.connection: A connection object to interact with the PostgreSQL database. 

18 ''' 

19 return psycopg2.connect(**self.__db_config) 

20 

21 def get_support_message(self, support_message: SupportMessageEntity) -> SupportMessageEntity: 

22 ''' 

23 Retrieves a support message from the PostgreSQL database by its entity. 

24 Args: 

25 support_message (SupportMessageEntity): The support message entity containing the ID to retrieve. 

26 Returns: 

27 SupportMessageEntity: The retrieved support message, or None if not found. 

28 Raises: 

29 psycopg2.Error: If an error occurs while retrieving the support message from the PostgreSQL database. 

30 ''' 

31 query = "SELECT * FROM Support WHERE id = %s;" 

32 with self.__connect() as conn: 

33 with conn.cursor() as cursor: 

34 cursor.execute(query, (support_message.get_id(),)) 

35 result = cursor.fetchone() 

36 if result: 

37 return SupportMessageEntity( 

38 id=result[0], 

39 user_id=result[1], 

40 description=result[2], 

41 status=result[3], 

42 subject=result[4], 

43 created_at=result[5] 

44 ) 

45 else: 

46 raise ValueError(f"No support messages found for support message ID {support_message.get_id()}.") 

47 

48 def get_support_messages(self) -> list[SupportMessageEntity]: 

49 ''' 

50 Retrieves all support messages from the PostgreSQL database, including user emails. 

51 Returns: 

52 list[SupportMessageEntity]: A list of all retrieved support messages with user emails. 

53 Raises: 

54 psycopg2.Error: If an error occurs while retrieving the support messages from the PostgreSQL database. 

55 ''' 

56 

57 query = """ 

58 SELECT s.id, s.user_id, u.email, s.description, s.status, s.subject, s.created_at 

59 FROM Support s 

60 LEFT JOIN Users u ON s.user_id = u.id; 

61 """ 

62 with self.__connect() as conn: 

63 with conn.cursor() as cursor: 

64 cursor.execute(query) 

65 results = cursor.fetchall() 

66 return [ 

67 SupportMessageEntity( 

68 id=row[0], 

69 user_id=row[1], 

70 user_email=row[2], 

71 description=row[3], 

72 status=row[4], 

73 subject=row[5], 

74 created_at=row[6] 

75 ) 

76 for row in results 

77 ] 

78 

79 

80 def save_support_message(self, support_message: SupportMessageEntity) -> int: 

81 ''' 

82 Saves a support message in the PostgreSQL database. 

83 Args: 

84 SupportMessageEntity (SupportMessageEntity): The support message entity to save. 

85 Returns: 

86 int: The ID of the saved support message. 

87 Raises: 

88 psycopg2.Error: If an error occurs while saving the support message in the PostgreSQL database. 

89 ''' 

90 

91 query = """ 

92 INSERT INTO Support (user_id, description, status, subject, created_at) 

93 VALUES (%s, %s, %s::boolean, %s, %s) 

94 RETURNING id; 

95 """ 

96 with self.__connect() as conn: 

97 with conn.cursor() as cursor: 

98 cursor.execute(query, (support_message.get_user_id(), support_message.get_description(), support_message.get_status(), support_message.get_subject(), support_message.get_created_at() )) 

99 conn.commit() 

100 return cursor.fetchone()[0] 

101 

102 def delete_support_message(self, support_message : SupportMessageEntity)-> bool: 

103 ''' 

104 Deletes a support message from the PostgreSQL database by its ID. 

105 Args: 

106 support_message (SupportMessageEntity): The support message entity to delete. 

107 Returns: 

108 bool: True if the support message was successfully deleted, False otherwise. 

109 Raises: 

110 psycopg2.Error: If an error occurs while deleting the support message from the PostgreSQL database. 

111 ''' 

112 query = "DELETE FROM Support WHERE id = %s;" 

113 with self.__connect() as conn: 

114 with conn.cursor() as cursor: 

115 cursor.execute(query, (support_message.get_id(),)) 

116 conn.commit() 

117 return cursor.rowcount > 0 

118 

119 

120 def mark_done_support_messages(self, support_message_model: SupportMessageEntity)-> int: 

121 ''' 

122 Marks a support message as done in the PostgreSQL database. 

123 Args: 

124 support_message_model (SupportMessageEntity): The support message entity to mark as done. 

125 Returns: 

126 int: The ID of the marked support message. 

127 Raises: 

128 psycopg2.Error: If an error occurs while marking the support message as done in the PostgreSQL database. 

129 ''' 

130 

131 query = """ 

132 UPDATE Support 

133 SET status = %s::boolean 

134 WHERE id = %s 

135 RETURNING id; 

136 """ 

137 with self.__connect() as conn: 

138 with conn.cursor() as cursor: 

139 cursor.execute(query, (support_message_model.get_status(), support_message_model.get_id())) 

140 conn.commit() 

141 return cursor.fetchone()[0] 

142 

143