wip

Author: m-bigaignon

entitled/client.py

@@ -1,28 +1,45 @@
 from typing import Any, Literal
 
 from entitled.exceptions import AuthorizationException
+from entitled.policies import Policy
 from entitled.response import Err, Response
 from entitled.rules import Actor, Rule, RuleProto
 
 
 class Client:
     def __init__(self):
+        self._policy_registry: dict[type, type[Policy[Any]]] = {}
         self._rule_registry: dict[str, Rule[Any]] = {}
 
     def define_rule(self, name: str, callable: RuleProto[Actor]) -> Rule[Actor]:
         rule = Rule(name, callable)
         self._rule_registry[rule.name] = rule
         return rule
 
+    def _register_policy(self, policy: type[Policy[Any]]):
+        resource_type = getattr(policy, "__orig_bases__")[0].__args__[0]
+        self._policy_registry[resource_type] = policy
+
+    def _resolve_rule(self, name: str, resource: Any) -> Rule[Any] | None:
+        lookup_key = resource if isinstance(resource, type) else type(resource)
+        policy = self._policy_registry.get(lookup_key, None)
+        if policy is not None:
+            rule = getattr(policy, name, None)
+            if rule is not None and hasattr(rule, "_is_rule"):
+                return Rule(name, rule)
+        return self._rule_registry.get(name, None)
+
     async def inspect(
         self,
         name: str,
         actor: Actor,
+        resource: Any,
         *args: Any,
         **kwargs: Any,
     ) -> Response:
-        if name in self._rule_registry:
-            return await self._rule_registry[name].inspect(actor, *args, **kwargs)
+        rule = self._resolve_rule(name, resource)
+        if rule is not None:
+            return await rule.inspect(actor, resource, *args, **kwargs)
         return Err(f"No rule found with name '{name}'")
 
     async def allows(

entitled/policies.py

@@ -1,11 +1,15 @@
-from typing import Any
-from entitled.rules import RuleProto
+import functools
+from typing import Any, Callable
 
 
-class Policy[T]:
-    def rule(self):
-        def wrapped(func: RuleProto[Any]):
-            func._is_rule = True
-            return func
+def rule(func: Callable[..., Any]):
+    @functools.wraps(func)
+    async def wrapper(*args: Any, **kwargs: Any) -> Any:
+        return await func(*args, **kwargs)
+
+    setattr(wrapper, "_is_rule", True)
+    return wrapper
 
-        return wrapped
+
+class Policy[T]:
+    pass

tests/test_policies.py

@@ -0,0 +1,43 @@
+import pytest
+
+from entitled.client import Client
+from entitled.policies import Policy, rule
+from entitled.response import Err, Ok, Response
+from tests.data.models import Tenant, User
+
+pytestmark = pytest.mark.anyio
+
+client = Client()
+
+
+class TenantPolicy(Policy[Tenant]):
+    @rule
+    async def is_member(
+        self,
+        actor: User,
+        resource: Tenant,
+    ) -> bool:
+        return actor.tenant == resource
+
+    @rule
+    async def is_owner(
+        self,
+        actor: User,
+        resource: Tenant,
+    ) -> Response:
+        return Ok() if resource.owner == actor else Err("Not owner on the tenant")
+
+
+async def test_register():
+    client._register_policy(TenantPolicy)
+    assert Tenant in client._policy_registry
+
+
+async def test_inspect():
+    tenant1 = Tenant("tenant1")
+    tenant2 = Tenant("tenant2")
+    user1 = User("user1", tenant1, set())
+    user2 = User("user2", None, set())
+    tenant2.owner = user2
+
+    assert await client.inspect("is_member", user1, tenant1)