Release v0.1.5

src/govoplan_access/backend/directory.py

@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+from collections.abc import Iterable, Mapping
+
+from govoplan_core.core.access import AccessDirectory, AccessSubjectRef, AccountRef, GroupRef, UserRef
+from govoplan_access.backend.db.models import Account, Group, User
+from govoplan_core.db.session import get_database
+
+
+def _status(active: bool) -> str:
+    return "active" if active else "inactive"
+
+
+def _account_ref(account: Account) -> AccountRef:
+    return AccountRef(
+        id=account.id,
+        email=account.email,
+        display_name=account.display_name,
+        status=_status(account.is_active),  # type: ignore[arg-type]
+    )
+
+
+def _user_ref(user: User, account: Account | None = None) -> UserRef:
+    return UserRef(
+        id=user.id,
+        account_id=user.account_id,
+        tenant_id=user.tenant_id,
+        email=account.email if account else user.email,
+        display_name=user.display_name or (account.display_name if account else None),
+        status=_status(user.is_active),  # type: ignore[arg-type]
+    )
+
+
+def _group_ref(group: Group) -> GroupRef:
+    return GroupRef(
+        id=group.id,
+        tenant_id=group.tenant_id,
+        name=group.name,
+        status=_status(group.is_active),  # type: ignore[arg-type]
+    )
+
+
+class SqlAccessDirectory(AccessDirectory):
+    def get_account(self, account_id: str) -> AccountRef | None:
+        with get_database().session() as session:
+            account = session.get(Account, account_id)
+            return _account_ref(account) if account is not None else None
+
+    def get_user(self, user_id: str) -> UserRef | None:
+        with get_database().session() as session:
+            user = session.get(User, user_id)
+            if user is None:
+                return None
+            account = session.get(Account, user.account_id)
+            return _user_ref(user, account)
+
+    def get_users(self, user_ids: Iterable[str]) -> Mapping[str, UserRef]:
+        ids = {str(user_id) for user_id in user_ids if user_id}
+        if not ids:
+            return {}
+        with get_database().session() as session:
+            users = session.query(User).filter(User.id.in_(ids)).all()
+            account_ids = {user.account_id for user in users}
+            accounts = {
+                account.id: account
+                for account in session.query(Account).filter(Account.id.in_(account_ids)).all()
+            } if account_ids else {}
+            return {user.id: _user_ref(user, accounts.get(user.account_id)) for user in users}
+
+    def users_for_tenant(self, tenant_id: str) -> tuple[UserRef, ...]:
+        with get_database().session() as session:
+            users = session.query(User).filter(User.tenant_id == tenant_id).order_by(User.display_name.asc(), User.email.asc()).all()
+            account_ids = {user.account_id for user in users}
+            accounts = {
+                account.id: account
+                for account in session.query(Account).filter(Account.id.in_(account_ids)).all()
+            } if account_ids else {}
+            return tuple(_user_ref(user, accounts.get(user.account_id)) for user in users)
+
+    def get_group(self, group_id: str) -> GroupRef | None:
+        with get_database().session() as session:
+            group = session.get(Group, group_id)
+            return _group_ref(group) if group is not None else None
+
+    def get_groups(self, group_ids: Iterable[str]) -> Mapping[str, GroupRef]:
+        ids = {str(group_id) for group_id in group_ids if group_id}
+        if not ids:
+            return {}
+        with get_database().session() as session:
+            groups = session.query(Group).filter(Group.id.in_(ids)).all()
+            return {group.id: _group_ref(group) for group in groups}
+
+    def groups_for_tenant(self, tenant_id: str) -> tuple[GroupRef, ...]:
+        with get_database().session() as session:
+            groups = session.query(Group).filter(Group.tenant_id == tenant_id).order_by(Group.name.asc()).all()
+            return tuple(_group_ref(group) for group in groups)
+
+    def groups_for_user(self, user_id: str, *, tenant_id: str) -> tuple[GroupRef, ...]:
+        with get_database().session() as session:
+            from govoplan_access.backend.db.models import UserGroupMembership
+
+            groups = (
+                session.query(Group)
+                .join(UserGroupMembership, UserGroupMembership.group_id == Group.id)
+                .filter(
+                    UserGroupMembership.user_id == user_id,
+                    UserGroupMembership.tenant_id == tenant_id,
+                    Group.tenant_id == tenant_id,
+                )
+                .order_by(Group.name.asc())
+                .all()
+            )
+            return tuple(_group_ref(group) for group in groups)
+
+    def display_label(self, subject: AccessSubjectRef) -> str | None:
+        if subject.kind == "account":
+            account = self.get_account(subject.id)
+            return account.display_name or account.email if account else subject.label
+        if subject.kind == "user":
+            user = self.get_user(subject.id)
+            return user.display_name or user.email if user else subject.label
+        if subject.kind == "group":
+            group = self.get_group(subject.id)
+            return group.name if group else subject.label
+        return subject.label or subject.id