💥 refactor Node

Author: RF-Tar-Railt

README.md

@@ -14,18 +14,29 @@ WIP
 ## Example
 
 ```python
-from arclet.cithun import User, NodeState, Group, context
+from arclet.cithun import SyncMonitor, Node, NodeState, context, PermissionExecutor
+
+monitor = SyncMonitor()
+
+baz = Node("/foo/bar/baz").mkdir(parents=True)
+qux = (baz / "qux").touch()
 
 with context(scope="main"):
-    admin = Group('admin', 100)
-    admin.add("/foo/bar/baz/", NodeState("vma"))
+    admin = monitor.new_group('admin', 100)
+    PermissionExecutor.root.set(admin, baz, NodeState("vma"))
 
-    user = User('cithun')
-    user.join(admin)
+    user = monitor.new_user('cithun')
+    monitor.user_inherit(user, admin)
 
-    user.get("/foo/bar/baz/")  # vma
-    user.sadd("/foo/bar/baz/qux")
-    user.available("/foo/bar/baz/qux")  # False as default perm of qux is vm-
-    admin.smodify("/foo/bar/baz/", NodeState("v-a"))
-    user.modify("/foo/bar/baz/qux", NodeState(7)) # False as /baz/ is not modifiable
+    assert PermissionExecutor.root.get(user, baz).most == NodeState("vma")
+    assert not PermissionExecutor.root.get(user, qux).most.available
+    
+    PermissionExecutor.root.set(user, qux, NodeState(7))
+    assert PermissionExecutor.root.get(user, qux).most.available
+
+    PermissionExecutor.root.set(admin, baz, NodeState("v-a"))
+    try:
+        PermissionExecutor(user).set(user, qux, NodeState("vm-"))
+    except PermissionError as e:
+        print(e)  # Permission denied as /baz/ is not modifiable
 ```

src/arclet/cithun/__init__.py

@@ -1,6 +1,8 @@
-from . import monitor as monitor
+from .function import PermissionExecutor as PermissionExecutor
 from .ctx import Context as Context
 from .ctx import context as context
+from .monitor import AsyncMonitor as AsyncMonitor
+from .monitor import SyncMonitor as SyncMonitor
 from .node import ROOT as ROOT
 from .node import Node as Node
 from .node import NodeState as NodeState

src/arclet/cithun/builtins/__init__.py

@@ -0,0 +1,76 @@
+import json
+from contextlib import contextmanager
+from pathlib import Path
+from weakref import WeakValueDictionary
+
+from arclet.cithun import Group, User
+from arclet.cithun.monitor import SyncMonitor
+
+from .owner import DefaultGroup, DefaultUser
+
+
+class DefaultMonitor(SyncMonitor):
+    def __init__(self, file: Path):
+        if not file.suffix.startswith(".json"):
+            raise ValueError(file)
+        self.file = file
+        self.USER_TABLE = WeakValueDictionary()
+        self.GROUP_TABLE = WeakValueDictionary()
+
+    def new_group(self, name: str, priority: int):
+        if name in self.GROUP_TABLE:
+            raise ValueError(f"Group {name} already exists")
+        group = DefaultGroup(name, priority)
+        self.GROUP_TABLE[name] = group
+        return group
+
+    def new_user(self, name: str):
+        if name in self.USER_TABLE:
+            raise ValueError(f"User {name} already exists")
+        user = DefaultUser(name)
+        self.USER_TABLE[name] = user
+        return user
+
+    def load(self):
+        if self.file.exists():
+            with self.file.open("r", encoding="utf-8") as f:
+                data = json.load(f)
+            users = {name: DefaultUser.parse(raw) for name, raw in data["users"].items()}
+            groups = {name: DefaultGroup.parse(raw) for name, raw in data["groups"].items()}
+            self.USER_TABLE.update(users)
+            self.GROUP_TABLE.update(groups)
+            for group in groups.values():
+                group.inherits = [self.GROUP_TABLE[gp.name] for gp in group.inherits]
+            for user in users.values():
+                user.inherits = [self.USER_TABLE[gp.name] for gp in user.inherits]
+            del users, groups
+        else:
+            with self.file.open("w+", encoding="utf-8") as f:
+                json.dump({}, f, ensure_ascii=False)
+
+    def save(self):
+        data = {
+            "users": {user.name: user.dump() for user in self.USER_TABLE.values()},
+            "groups": {group.name: group.dump() for group in self.GROUP_TABLE.values()},
+        }
+        with self.file.open("w", encoding="utf-8") as f:
+            json.dump(data, f, ensure_ascii=False)
+
+    def group_inherit(self, target: Group, *groups: Group):
+        for group in groups:
+            if group not in target.inherits:
+                target.inherits.append(group)
+
+    def user_inherit(self, target: User, *groups: Group):
+        for group in groups:
+            if group not in target.inherits:
+                target.inherits.append(group)
+
+    def user_leave(self, target: User, group: Group):
+        if group in target.inherits:
+            target.inherits.remove(group)
+
+    @contextmanager
+    def transaction(self):
+        yield
+        self.save()

src/arclet/cithun/builtins/monitor.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from arclet.cithun.ctx import Context
+from arclet.cithun.node import Node, NodeState
+
+
+@dataclass(eq=True, unsafe_hash=True)
+class DefaultGroup:
+    name: str
+    priority: int
+    nodes: dict[Node, dict[Context, NodeState]] = field(default_factory=dict, compare=False, hash=False)
+    inherits: list = field(default_factory=list, compare=False, hash=False)
+
+    def dump(self):
+        return {
+            "name": self.name,
+            "priority": self.priority,
+            "nodes": {
+                str(node): {str(ctx): state.state for ctx, state in data.items()} for node, data in self.nodes.items()
+            },
+            "inherits": [gp.name for gp in self.inherits],
+        }
+
+    @classmethod
+    def parse(cls, raw: dict):
+        obj = cls(raw["name"], raw["priority"])
+        obj.nodes = {
+            Node(node): {Context.from_string(ctx): NodeState(state) for ctx, state in data.items()}
+            for node, data in raw["nodes"].items()
+        }
+        obj.inherits = [DefaultGroup(name, 0) for name in raw["inherits"]]
+        return obj
+
+    def __str__(self):
+        return f"Group({self.name})"
+
+
+@dataclass(eq=True, unsafe_hash=True)
+class DefaultUser:
+    name: str
+    nodes: dict[Node, dict[Context, NodeState]] = field(default_factory=dict, compare=False, hash=False)
+    inherits: list = field(default_factory=list, compare=False, hash=False)
+
+    def dump(self):
+        return {
+            "name": self.name,
+            "nodes": {
+                str(node): {str(ctx): state.state for ctx, state in data.items()} for node, data in self.nodes.items()
+            },
+            "inherits": [user.name for user in self.inherits],
+        }
+
+    @classmethod
+    def parse(cls, raw: dict):
+        obj = cls(raw["name"])
+        obj.nodes = {
+            Node(node): {Context.from_string(ctx): NodeState(state) for ctx, state in data.items()}
+            for node, data in raw["nodes"].items()
+        }
+        obj.inherits = [DefaultUser(name) for name in raw["inherits"]]
+        return obj
+
+    def __str__(self):
+        return f"User({self.name})"

src/arclet/cithun/builtins/owner.py

@@ -1,7 +1,7 @@
 from collections import UserDict
 from contextlib import contextmanager
 from contextvars import ContextVar
-from typing import Literal
+from typing import Callable, Dict, Generic, TypeVar
 
 
 class Context(UserDict):
@@ -29,19 +29,6 @@ def from_string(cls, data: str):
                 _data[segs[0]] = segs[1]
         return cls(**_data)
 
-    def satisfied(self, other: "Context", mode: Literal["least_one", "all"] = "least_one") -> bool:
-        if not self.data:
-            return True
-        if not other.data:
-            return True
-        return self.contain_all(other) if mode == "all" else self.contain_least(other)
-
-    def contain_all(self, other: "Context"):
-        return self.data == other.data
-
-    def contain_least(self, other: "Context"):
-        return any((other.get(k) == v for k, v in self.data.items()))
-
 
 _ctx = ContextVar("context")
 
@@ -53,3 +40,44 @@ def context(**kwargs):
         yield
     finally:
         _ctx.reset(token)
+
+
+class Satisfier:
+    @staticmethod
+    def all():
+        return Satisfier(lambda self, other: self.data == other.data)
+
+    @staticmethod
+    def least():
+        return Satisfier(lambda self, other: any((other.get(k) == v for k, v in self.data.items())))
+
+    def __init__(self, func: Callable[[Context, Context], bool]):
+        self.func = func
+
+    def __call__(self, target: "Context", other: "Context") -> bool:
+        if not target.data:
+            return True
+        if not other.data:
+            return True
+        return self.func(target, other)
+
+
+T = TypeVar("T")
+
+
+class Result(Generic[T]):
+    def __init__(self, data: Dict[Context, T], origin: Context):
+        self.data = data
+        self.origin = origin
+
+    @property
+    def most(self) -> T:
+        if self.origin in self.data:
+            return self.data[self.origin]
+        sortd = sorted(
+            self.data.keys(), key=lambda x: sum((x.get(k) == v for k, v in self.origin.items())), reverse=True
+        )
+        return self.data[sortd[0]]
+
+    def __repr__(self):
+        return f"Result({self.data})"