Coverage for src/app/adapters/conversation_postgres_adapter.py: 39%

1from repositories.conversation_postgres_repository import ConversationPostgresRepository 

2from models.conversation_model import ConversationModel 

3 

4from ports.save_conversation_title_port import SaveConversationTitlePort 

5from ports.get_conversations_port import GetConversationsPort 

6from ports.get_conversation_port import GetConversationPort 

7from ports.delete_conversation_port import DeleteConversationPort 

8 

9from entities.conversation_entity import ConversationEntity 

10 

11class ConversationPostgresAdapter(GetConversationPort, GetConversationsPort, SaveConversationTitlePort, DeleteConversationPort): 

12 

13 def __init__(self, conversation_postgres_repository: ConversationPostgresRepository): 

14 self.conversation_postgres_repository = conversation_postgres_repository 

15 

16 def get_conversation(self, conversation: ConversationModel) -> ConversationModel: 

17 """ 

18 Retrieve a conversation. 

19 Args: 

20 conversation (ConversationModel): The conversation object containing the ID to retrieve. 

21 Returns: 

22 ConversationModel: The retrieved conversation. 

23 """ 

24 try: 

25 

26 conversation_entity = ConversationEntity( 

27 id=conversation.get_id(), 

28 title=conversation.get_title(), 

29 user_id=conversation.get_user_id() 

30 ) 

31 

32 conversation = self.conversation_postgres_repository.get_conversation(conversation_entity) 

33 

34 return ConversationModel( 

35 id=conversation.get_id(), 

36 title=conversation.get_title(), 

37 user_id=conversation.get_user_id() 

38 ) 

39 

40 except Exception as e: 

41 raise e 

42 

43 def get_conversations(self, conversation : ConversationModel) -> list[ConversationModel]: 

44 """ 

45 Retrieve all conversations. 

46 Returns: 

47 list[ConversationModel]: A list of conversations. 

48 """ 

49 try: 

50 

51 conversation_entity = ConversationEntity( 

52 id=conversation.get_id(), 

53 title=conversation.get_title(), 

54 user_id=conversation.get_user_id() 

55 ) 

56 

57 conversations = self.conversation_postgres_repository.get_conversations(conversation_entity) 

58 

59 return [ 

60 ConversationModel( 

61 id=conversation.get_id(), 

62 title=conversation.get_title(), 

63 user_id=conversation.get_user_id() 

64 ) 

65 for conversation in conversations 

66 ] 

67 

68 except Exception as e: 

69 raise e 

70 

71 def save_conversation_title(self, conversation: ConversationModel) -> int: 

72 """ 

73 Save the title of a conversation. 

74 Args: 

75 conversation (ConversationModel): The conversation object containing the updated title. 

76 Returns: 

77 int: The ID of the saved conversation. 

78 """ 

79 try: 

80 

81 conversation_entity = ConversationEntity( 

82 id=conversation.get_id(), 

83 title=conversation.get_title(), 

84 user_id=conversation.get_user_id() 

85 ) 

86 

87 return self.conversation_postgres_repository.save_conversation_title(conversation_entity) 

88 

89 

90 except Exception as e: 

91 raise e 

92 

93 def delete_conversation_title(self, conversation: ConversationModel) -> bool: 

94 """ 

95 Delete a conversation title. 

96 Args: 

97 conversation (ConversationModel): The conversation object containing the ID to delete. 

98 Returns: 

99 int: The ID of the deleted conversation. 

100 """ 

101 try: 

102 

103 conversation_entity = ConversationEntity( 

104 id=conversation.get_id(), 

105 title=conversation.get_title(), 

106 user_id=conversation.get_user_id() 

107 ) 

108 

109 return self.conversation_postgres_repository.delete_conversation(conversation_entity) 

110 

111 

112 except Exception as e: 

113 raise e