Coverage for src / lexigram / contracts / auth / policy.py: 100%

1"""Policy-store protocol for the ABAC policy engine. 

2 

3Defines :class:`PolicyStoreProtocol`, the contract for loading, persisting, and 

4querying ABAC policies. Multiple back-ends (in-memory, YAML file, 

5database, remote service) can satisfy this protocol through structural 

6subtyping, making the :class:`PolicyEngine` independently testable. 

7""" 

8 

9from __future__ import annotations 

10 

11from typing import Any, Protocol, runtime_checkable 

12 

13__all__ = ["PolicyStoreProtocol"] 

14 

15 

16@runtime_checkable 

17class PolicyStoreProtocol(Protocol): 

18 """Storage protocol for ABAC authorization policies. 

19 

20 Implementations are responsible for persisting and retrieving 

21 :class:`~lexigram.auth.policies.types.Policy` objects. The 

22 concrete ``Policy`` type is referenced as ``Any`` here so that the 

23 contracts package remains dependency-free. 

24 

25 Example:: 

26 

27 class YAMLPolicyStore: 

28 def __init__(self, path: Path) -> None: 

29 self._path = path 

30 

31 async def load_policies(self) -> list[Any]: 

32 with open(self._path) as f: 

33 raw = yaml.safe_load(f) 

34 return [Policy(**p) for p in raw["policies"]] 

35 

36 async def save_policy(self, policy: Any) -> None: 

37 ... # Append/update in YAML file 

38 """ 

39 

40 async def load_policies(self) -> list[Any]: 

41 """Load all policies from the store. 

42 

43 Returns: 

44 List of policy objects ordered by their natural storage order. 

45 """ 

46 ... 

47 

48 async def save_policy(self, policy: Any) -> None: 

49 """Persist a single policy. 

50 

51 Args: 

52 policy: The policy object to save. Implementations should 

53 generate the ``id`` field if it is absent. 

54 """ 

55 ... 

56 

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

58 """Remove a policy by its identifier. 

59 

60 Args: 

61 policy_id: Unique policy identifier. 

62 

63 Returns: 

64 True if the policy was found and removed, False otherwise. 

65 """ 

66 ... 

67 

68 async def get_policy(self, policy_id: str) -> Any | None: 

69 """Retrieve a single policy by ID. 

70 

71 Args: 

72 policy_id: Unique policy identifier. 

73 

74 Returns: 

75 The policy if found, None otherwise. 

76 """ 

77 ...