"""Identity & Access domain: users, organizations, memberships, invites.""" from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime from enum import StrEnum from app.domain.base import new_id, utcnow class Role(StrEnum): OWNER = "owner" ADMIN = "admin" MANAGER = "manager" MEMBER = "member" VIEWER = "viewer" #: Higher rank implies every permission of the lower ranks. ROLE_RANK: dict[Role, int] = { Role.VIEWER: 10, Role.MEMBER: 20, Role.MANAGER: 30, Role.ADMIN: 40, Role.OWNER: 50, } #: Capabilities used across the application layer. CAN_MANAGE_ORG = "org.manage" CAN_MANAGE_MEMBERS = "org.members" CAN_MANAGE_PROJECTS = "projects.manage" CAN_MANAGE_EMPLOYEES = "employees.manage" CAN_RUN_TASKS = "tasks.run" CAN_VIEW_BILLING = "billing.view" CAN_MANAGE_BILLING = "billing.manage" CAN_VIEW_AUDIT = "audit.view" CAN_USE_SANDBOX = "sandbox.use" _ROLE_CAPABILITIES: dict[Role, set[str]] = { Role.VIEWER: set(), Role.MEMBER: {CAN_RUN_TASKS, CAN_VIEW_BILLING}, Role.MANAGER: { CAN_RUN_TASKS, CAN_VIEW_BILLING, CAN_MANAGE_PROJECTS, CAN_MANAGE_EMPLOYEES, CAN_USE_SANDBOX, }, Role.ADMIN: { CAN_RUN_TASKS, CAN_VIEW_BILLING, CAN_MANAGE_BILLING, CAN_MANAGE_PROJECTS, CAN_MANAGE_EMPLOYEES, CAN_USE_SANDBOX, CAN_MANAGE_MEMBERS, CAN_VIEW_AUDIT, }, Role.OWNER: { CAN_RUN_TASKS, CAN_VIEW_BILLING, CAN_MANAGE_BILLING, CAN_MANAGE_PROJECTS, CAN_MANAGE_EMPLOYEES, CAN_USE_SANDBOX, CAN_MANAGE_MEMBERS, CAN_VIEW_AUDIT, CAN_MANAGE_ORG, }, } def role_has_capability(role: Role, capability: str) -> bool: return capability in _ROLE_CAPABILITIES.get(role, set()) class MembershipStatus(StrEnum): ACTIVE = "active" INVITED = "invited" SUSPENDED = "suspended" @dataclass(slots=True, kw_only=True) class User: email: str password_hash: str name: str = "" id: str = field(default_factory=new_id) avatar_url: str | None = None is_platform_owner: bool = False is_active: bool = True created_at: datetime = field(default_factory=utcnow) updated_at: datetime = field(default_factory=utcnow) @property def display_name(self) -> str: return self.name.strip() or self.email.split("@")[0] def normalized_email(self) -> str: return self.email.strip().lower() @dataclass(slots=True, kw_only=True) class Organization: name: str slug: str owner_user_id: str id: str = field(default_factory=new_id) created_at: datetime = field(default_factory=utcnow) @dataclass(slots=True, kw_only=True) class Membership: organization_id: str user_id: str role: Role = Role.MEMBER status: MembershipStatus = MembershipStatus.ACTIVE id: str = field(default_factory=new_id) created_at: datetime = field(default_factory=utcnow) @property def is_active(self) -> bool: return self.status == MembershipStatus.ACTIVE def has(self, capability: str) -> bool: return self.is_active and role_has_capability(self.role, capability) def outranks(self, other_role: Role) -> bool: return ROLE_RANK[self.role] > ROLE_RANK[other_role] @dataclass(slots=True, kw_only=True) class Invite: organization_id: str email: str role: Role token_hash: str expires_at: datetime id: str = field(default_factory=new_id) invited_by_user_id: str | None = None accepted_at: datetime | None = None created_at: datetime = field(default_factory=utcnow) def is_expired(self, now: datetime | None = None) -> bool: return (now or utcnow()) >= self.expires_at def is_pending(self, now: datetime | None = None) -> bool: return self.accepted_at is None and not self.is_expired(now)