Coverage for src/lexigram/auth/storage/oauth_identity_store/_protocol.py: 74%

1"""OAuth identity store protocol and identity model.""" 

2 

3from __future__ import annotations 

4 

5from datetime import datetime 

6from typing import Protocol, runtime_checkable 

7 

8 

9@runtime_checkable 

10class OAuthIdentityStore(Protocol): 

11 """Protocol for OAuth identity storage. 

12 

13 OAuthIdentityStore manages the linking between local user accounts and 

14 external OAuth2 identity providers. This enables users to authenticate 

15 using social login while maintaining a consistent local user identity. 

16 

17 Example: 

18 Using OAuth identity store:: 

19 

20 store = await container.resolve(OAuthIdentityStore) 

21 

22 # Link OAuth identity to user 

23 identity = await store.create_oauth_identity( 

24 user_id="user-123", 

25 provider="google", 

26 provider_user_id="google-456" 

27 ) 

28 

29 # Find user by OAuth identity 

30 identity = await store.get_oauth_identity("google", "google-456") 

31 if identity: 

32 user = await store.get_user_by_oauth_identity(identity.id) 

33 """ 

34 

35 async def create_oauth_identity( 

36 self, 

37 user_id: str, 

38 provider: str, 

39 provider_user_id: str, 

40 ) -> OAuthIdentity: 

41 """Create an OAuth identity link for a user. 

42 

43 Args: 

44 user_id: The local user ID to link the OAuth identity to. 

45 provider: The OAuth provider name (e.g., "google", "github"). 

46 provider_user_id: The user's ID at the OAuth provider. 

47 

48 Returns: 

49 The created OAuthIdentity instance. 

50 """ 

51 ... 

52 

53 async def get_oauth_identity( 

54 self, 

55 provider: str, 

56 provider_user_id: str, 

57 ) -> OAuthIdentity | None: 

58 """Get OAuth identity by provider and provider user ID. 

59 

60 Args: 

61 provider: The OAuth provider name. 

62 provider_user_id: The user's ID at the OAuth provider. 

63 

64 Returns: 

65 The OAuthIdentity if found, None otherwise. 

66 """ 

67 ... 

68 

69 async def get_oauth_identities_for_user( 

70 self, 

71 user_id: str, 

72 ) -> list[OAuthIdentity]: 

73 """Get all OAuth identities for a user""" 

74 ... 

75 

76 async def delete_oauth_identity( 

77 self, 

78 provider: str, 

79 provider_user_id: str, 

80 ) -> bool: 

81 """Delete OAuth identity""" 

82 ... 

83 

84 async def delete_oauth_identities_for_user(self, user_id: str) -> int: 

85 """Delete all OAuth identities for a user""" 

86 ... 

87 

88 async def get_user_by_oauth_identity( 

89 self, 

90 provider: str, 

91 provider_user_id: str, 

92 ) -> str | None: 

93 """Get local user_id by OAuth provider and external user ID. 

94 

95 This is the key method for resolving OAuth external IDs to local user IDs. 

96 Used when OAuth tokens contain non-UUID user identifiers (like Google's sub claim). 

97 

98 Args: 

99 provider: The OAuth provider name (e.g., "google", "github", "apple"). 

100 provider_user_id: The user's ID at the OAuth provider. 

101 

102 Returns: 

103 The local user_id if found, None otherwise. 

104 """ 

105 ... 

106 

107 async def resolve_user_id( 

108 self, 

109 user_id_or_oauth_id: str, 

110 provider: str = "google", 

111 ) -> str | None: 

112 """Resolve a user_id that may be either a UUID or an OAuth external ID. 

113 

114 This method handles the common issue where OAuth providers (like Google) 

115 use non-UUID identifiers (e.g., "101158382316025899191") which cannot 

116 be used directly in database queries expecting UUIDs. 

117 

118 Resolution logic: 

119 1. If user_id_or_oauth_id is a valid UUID format, check if user exists 

120 2. If not a valid UUID, treat it as an OAuth provider_user_id and look up 

121 

122 Args: 

123 user_id_or_oauth_id: Either a local user UUID or an OAuth provider's external ID. 

124 provider: The OAuth provider to search in (default: "google"). 

125 

126 Returns: 

127 The resolved local user_id if found, None otherwise. 

128 """ 

129 ... 

130 

131 

132class OAuthIdentity: 

133 """OAuth identity linking user to provider""" 

134 

135 def __init__( 

136 self, 

137 user_id: str, 

138 provider: str, 

139 provider_user_id: str, 

140 created_at: datetime | None = None, 

141 updated_at: datetime | None = None, 

142 ): 

143 self.user_id = user_id 

144 self.provider = provider 

145 self.provider_user_id = provider_user_id 

146 self.created_at = created_at 

147 self.updated_at = updated_at