Coverage for src/lexigram/auth/policies/store.py: 100%

1"""Policy store protocols for ABAC policy persistence.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Protocol, runtime_checkable 

6 

7if TYPE_CHECKING: 

8 from lexigram.auth.policies.types import Policy 

9 

10 

11@runtime_checkable 

12class PolicyStoreProtocol(Protocol): 

13 """Protocol for policy storage backends. 

14 

15 Implement this protocol to provide custom policy storage 

16 (database, file, remote service, etc.). 

17 """ 

18 

19 async def get_all(self) -> list[Policy]: 

20 """Get all policies from the store. 

21 

22 Returns: 

23 List of all policies. 

24 """ 

25 ... 

26 

27 async def get_by_id(self, policy_id: str) -> Policy | None: 

28 """Get a policy by its ID. 

29 

30 Args: 

31 policy_id: The policy ID. 

32 

33 Returns: 

34 The policy if found, None otherwise. 

35 """ 

36 ... 

37 

38 async def get_by_name(self, name: str) -> Policy | None: 

39 """Get a policy by its name. 

40 

41 Args: 

42 name: The policy name. 

43 

44 Returns: 

45 The policy if found, None otherwise. 

46 """ 

47 ... 

48 

49 async def save(self, policy: Policy) -> None: 

50 """Save a policy to the store. 

51 

52 Args: 

53 policy: The policy to save. 

54 """ 

55 ... 

56 

57 async def delete(self, policy_id: str) -> bool: 

58 """Delete a policy from the store. 

59 

60 Args: 

61 policy_id: The policy ID to delete. 

62 

63 Returns: 

64 True if deleted, False if not found. 

65 """ 

66 ... 

67 

68 async def exists(self, policy_id: str) -> bool: 

69 """Check if a policy exists. 

70 

71 Args: 

72 policy_id: The policy ID. 

73 

74 Returns: 

75 True if exists, False otherwise. 

76 """ 

77 ... 

78 

79 

80@runtime_checkable 

81class PolicyLoader(Protocol): 

82 """Protocol for loading policies from various sources. 

83 

84 Used to load policies from files, databases, or remote services. 

85 """ 

86 

87 async def load(self) -> list[Policy]: 

88 """Load policies from the source. 

89 

90 Returns: 

91 List of loaded policies. 

92 """ 

93 ... 

94 

95 async def reload(self) -> list[Policy]: 

96 """Reload policies from the source. 

97 

98 Returns: 

99 List of reloaded policies. 

100 """ 

101 ... 

102 

103 

104__all__ = [ 

105 "PolicyLoader", 

106 "PolicyStoreProtocol", 

107]