Coverage for functions \ flipdare \ firestore \ friend_db.py: 51%

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

4# This file is part of Flipdare's proprietary software and contains 

5# confidential and copyrighted material. Unauthorised copying, 

6# modification, distribution, or use of this file is strictly 

7# prohibited without prior written permission from Flipdare Pty Ltd. 

8# 

9# This software includes third-party components licensed under MIT, 

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

11# 

12 

13 

14from google.cloud.firestore import Client as FirestoreClient 

15from flipdare.app_log import LOG 

16from flipdare.constants import IS_DEBUG 

17from flipdare.error.app_error import DatabaseError 

18from flipdare.firestore._app_db import AppDb 

19from flipdare.firestore.core.db_query import DbQuery, FieldOp, WhereField 

20from flipdare.generated import FriendKeys 

21from flipdare.generated.model.friend_model import FriendInternalKeys, FriendModel 

22from flipdare.generated.shared.app_error_code import AppErrorCode 

23from flipdare.generated.shared.firestore_collections import FirestoreCollections 

24from flipdare.util.time_util import TimeUtil 

25from flipdare.wrapper import FriendWrapper 

26 

27_FRIEND = FirestoreCollections.FRIEND.value 

28 

29__all__ = ["FriendDb"] 

30 

31_K = FriendKeys 

32_I = FriendInternalKeys 

33 

34 

35class FriendDb(AppDb[FriendWrapper, FriendModel]): 

36 """Class for managing friend-related database operations.""" 

37 

38 def __init__(self, client: FirestoreClient) -> None: 

39 super().__init__( 

40 client=client, 

41 collection_name=FirestoreCollections.FRIEND, 

42 model_class=FriendModel, 

43 wrapper_class=FriendWrapper, 

44 ) 

45 

46 def get_friend(self, from_uid: str, to_uid: str) -> FriendWrapper | None: 

47 where_query = DbQuery.and_( 

48 where_fields=[ 

49 WhereField[_K](_K.FROM_UID, FieldOp.EQUAL, from_uid), 

50 WhereField[_K](_K.TO_UID, FieldOp.EQUAL, to_uid), 

51 ], 

52 limit=1, 

53 ) 

54 

55 results = where_query.get_query(self.client, _FRIEND).get() 

56 if not results or len(results) == 0: 

57 return None 

58 

59 return self._cvt_snap_to_model(results[0]) 

60 

61 def get_unprocessed_friends_last_week(self) -> list[FriendWrapper]: 

62 """Get friends from the last 7 days.""" 

63 LOG().debug("Getting invites from the last week") 

64 try: 

65 one_week_ago = TimeUtil.get_utc_time_days_ago(7) 

66 query = DbQuery.and_( 

67 where_fields=[ 

68 WhereField[_I](_I.CREATED_AT, FieldOp.GREATER_THAN_OR_EQUAL, one_week_ago), 

69 WhereField[_I](_I.PROCESSED, FieldOp.EQUAL, False), 

70 ], 

71 ) 

72 results = query.get_query(self.client, _FRIEND).get() 

73 

74 if len(results) == 0: 

75 LOG().debug("No invites found from the last week") 

76 return [] 

77 

78 friends = [ 

79 friend for doc in results if (friend := self._cvt_snap_to_model(doc)) is not None 

80 ] 

81 if IS_DEBUG: 

82 LOG().debug(f"Retrieved {len(friends)} friends that can be processed.") 

83 

84 return friends 

85 except Exception as error: 

86 msg = "Error getting friends from last week" 

87 raise DatabaseError( 

88 msg, 

89 error_code=AppErrorCode.DATABASE_EX, 

90 collection_name=_FRIEND, 

91 ) from error