Coverage for src/lexigram/admin/auth/store/database.py: 33%

1""" 

2Database-backed admin user store implementation. 

3""" 

4 

5from __future__ import annotations 

6 

7from typing import TYPE_CHECKING 

8 

9from lexigram.admin.auth.store.base import AbstractAdminUserStore 

10from lexigram.admin.exceptions import AdminDataError 

11from lexigram.logging import get_logger 

12 

13if TYPE_CHECKING: 

14 from lexigram.admin.auth.entity import AdminUserEntity 

15 from lexigram.contracts import AuthenticatedUserProtocol, RepositoryProtocol 

16 from lexigram.contracts.auth import PasswordHasherProtocol 

17from lexigram.di.decorators import inject 

18 

19logger = get_logger(__name__) 

20 

21 

22@inject 

23class DatabaseAdminUserStore(AbstractAdminUserStore): 

24 """Database-backed store for admin users. 

25 

26 Uses a repository to fetch and authenticate users from admin_users table. 

27 """ 

28 

29 def __init__( 

30 self, 

31 repository: RepositoryProtocol[AdminUserEntity], 

32 hasher: PasswordHasherProtocol | None = None, 

33 ): 

34 """Initialize with repository. 

35 

36 Args: 

37 repository: RepositoryProtocol for AdminUserEntity 

38 hasher: Optional password hasher for authentication 

39 """ 

40 self.repo = repository 

41 self._hasher = hasher 

42 

43 async def get_by_id(self, user_id: str) -> AuthenticatedUserProtocol | None: 

44 """Get user by id via repository and convert to `User`.""" 

45 try: 

46 entity = await self.repo.find_by_id(user_id) # type: ignore[attr-defined] 

47 if entity is None: 

48 return None 

49 # Convert DB entity to User dataclass 

50 if hasattr(entity, "to_user"): 

51 return entity.to_user() 

52 return entity 

53 except (ConnectionError, RuntimeError, ValueError, OSError) as e: 

54 raise AdminDataError(f"Failed to retrieve user {user_id}: {e}") from e 

55 

56 async def get_by_email(self, email: str) -> AuthenticatedUserProtocol | None: 

57 """Get user by email via repository and convert to `User`.""" 

58 try: 

59 entity = await self.repo.find_one(email=email) # type: ignore[attr-defined] 

60 if entity is None: 

61 return None 

62 if hasattr(entity, "to_user"): 

63 return entity.to_user() 

64 return entity 

65 except (ConnectionError, RuntimeError, ValueError, OSError) as e: 

66 raise AdminDataError(f"Failed to find user by email: {e}") from e 

67 

68 async def get_by_username(self, username: str) -> AuthenticatedUserProtocol | None: 

69 """Get user by username via repository and convert to `User`.""" 

70 try: 

71 entity = await self.repo.find_one(username=username) # type: ignore[attr-defined] 

72 if entity is None: 

73 return None 

74 if hasattr(entity, "to_user"): 

75 return entity.to_user() 

76 return entity 

77 except (ConnectionError, RuntimeError, ValueError, OSError) as e: 

78 raise AdminDataError(f"Failed to find user by username: {e}") from e 

79 

80 async def authenticate( 

81 self, email: str, password: str 

82 ) -> AuthenticatedUserProtocol | None: 

83 """Authenticate user by email/password using DB repository.""" 

84 user = await self.get_by_email(email) 

85 if not user: 

86 return None 

87 

88 if not getattr(user, "is_active", True): 

89 return None 

90 

91 hashed = getattr(user, "hashed_password", None) 

92 if not hashed: 

93 return None 

94 

95 try: 

96 if self._hasher: 

97 if await self._hasher.verify(password, hashed): 

98 return user 

99 else: 

100 import hashlib 

101 

102 if hashlib.sha256(password.encode()).hexdigest() == hashed: 

103 return user 

104 except (ValueError, RuntimeError, OSError) as e: 

105 raise AdminDataError( 

106 f"Authentication verification failed for {email}: {e}" 

107 ) from e 

108 

109 return None 

110 

111 async def count(self) -> int: 

112 """Return total users via repo.count().""" 

113 try: 

114 return await self.repo.count() 

115 except (ConnectionError, RuntimeError, OSError) as e: 

116 raise AdminDataError(f"Failed to count users: {e}") from e