Coverage for src/app/repositories/user_postgres_repository.py: 57%

1import psycopg2 

2 

3from entities.user_entity import UserEntity 

4 

5 

6class UserPostgresRepository: 

7 

8 def __init__(self, db_config: dict): 

9 ''' 

10 Initializes the PostgresRepository with the given database configuration. 

11 Args: 

12 db_config (dict): The configuration dictionary for the PostgreSQL database. 

13 ''' 

14 self.__db_config = db_config 

15 

16 def __connect(self): 

17 ''' 

18 Establishes a new connection to the PostgreSQL database. 

19 Returns: 

20 psycopg2.extensions.connection: The connection object to the PostgreSQL database. 

21 ''' 

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

23 

24 def register(self, user_model: UserEntity)-> bool: 

25 """ 

26 Register a new user. 

27  

28 Args: 

29 user_model (UserModel): The user data transfer object. 

30  

31 Returns: 

32 bool: True if the user was registered successfully, False otherwise. 

33 """ 

34 

35 query = "INSERT INTO Users (username, password_hash, email, phone, first_name, last_name, is_admin) VALUES (%s, %s, %s, %s, %s, %s, %s);" 

36 with self.__connect() as conn: 

37 with conn.cursor() as cursor: 

38 cursor.execute(query, (user_model.get_username(), user_model.get_password(), user_model.get_email(), user_model.get_phone(), user_model.get_first_name(), user_model.get_last_name(), user_model.get_is_admin())) 

39 conn.commit() 

40 return True 

41 

42 def get_user_by_email(self, email: str) -> bool: 

43 """ 

44 Get a user by email. 

45  

46 Args: 

47 email (str): The email of the user. 

48  

49 Returns: 

50 bool: True if the user exists, False otherwise. 

51 """ 

52 query = "SELECT * FROM Users WHERE email = %s;" 

53 with self.__connect() as conn: 

54 with conn.cursor() as cursor: 

55 cursor.execute(query, (email,)) 

56 result = cursor.fetchone() 

57 if result: 

58 return True 

59 else: 

60 return False 

61 

62 def get_user_by_username(self, username: str) -> bool: 

63 """ 

64 Get a user by username. 

65  

66 Args: 

67 username (str): The username of the user. 

68  

69 Returns: 

70 bool: True if the user exists, False otherwise. 

71 """ 

72 query = "SELECT * FROM Users WHERE username = %s;" 

73 with self.__connect() as conn: 

74 with conn.cursor() as cursor: 

75 cursor.execute(query, (username,)) 

76 result = cursor.fetchone() 

77 if result: 77 ↛ 80line 77 didn't jump to line 80 because the condition on line 77 was always true

78 return True 

79 else: 

80 return False 

81 

82 def get_user_for_authentication(self, user : UserEntity) -> UserEntity: 

83 """ 

84 Authenticate and retrieve a user. 

85  

86 Args: 

87 user (UserModel): The user model to authenticate. 

88 

89 Returns: 

90 UserModel: The authenticated user model. 

91 """ 

92 

93 query = "SELECT * FROM Users WHERE username = %s;" 

94 with self.__connect() as conn: 

95 with conn.cursor() as cursor: 

96 cursor.execute(query, (user.get_username(),)) 

97 result = cursor.fetchone() 

98 if result: 98 ↛ 110line 98 didn't jump to line 110 because the condition on line 98 was always true

99 return UserEntity( 

100 id=result[0], 

101 username=result[1], 

102 password=result[2], 

103 email=result[3], 

104 phone=result[4], 

105 first_name=result[5], 

106 last_name=result[6], 

107 is_admin=result[7] 

108 ) 

109 else: 

110 return None 

111 

112