"""User model with RBAC support.""" from datetime import datetime from enum import Enum from uuid import UUID, uuid4 from sqlalchemy import String, ForeignKey, Integer, DateTime, Boolean, Table, Column from sqlalchemy.orm import Mapped, mapped_column, relationship from app.database import Base class UserStatus(str, Enum): ACTIVE = "active" INACTIVE = "inactive" PENDING = "pending" LOCKED = "locked" # Association table for user roles user_roles = Table( "user_roles", Base.metadata, Column("user_id", ForeignKey("users.id", ondelete="CASCADE"), primary_key=True), Column("role_id", ForeignKey("roles.id", ondelete="CASCADE"), primary_key=True), Column("assigned_by", ForeignKey("users.id")), Column("assigned_at", DateTime, default=datetime.utcnow), ) class User(Base): __tablename__ = "users" id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4) tenant_id: Mapped[UUID] = mapped_column( ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False ) email: Mapped[str] = mapped_column(String(255), nullable=False) password_hash: Mapped[str] = mapped_column(String(255), nullable=False) first_name: Mapped[str | None] = mapped_column(String(100)) last_name: Mapped[str | None] = mapped_column(String(100)) phone: Mapped[str | None] = mapped_column(String(50)) avatar_url: Mapped[str | None] = mapped_column(String(500)) status: Mapped[UserStatus] = mapped_column(default=UserStatus.ACTIVE) last_login_at: Mapped[datetime | None] failed_login_attempts: Mapped[int] = mapped_column(Integer, default=0) locked_until: Mapped[datetime | None] mfa_enabled: Mapped[bool] = mapped_column(Boolean, default=False) mfa_secret: Mapped[str | None] = mapped_column(String(255)) created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow) updated_at: Mapped[datetime] = mapped_column( default=datetime.utcnow, onupdate=datetime.utcnow ) # Relationships tenant: Mapped["Tenant"] = relationship(back_populates="users") roles: Mapped[list["Role"]] = relationship( secondary=user_roles, back_populates="users" ) audit_logs: Mapped[list["AuditLog"]] = relationship(back_populates="user") api_keys_created: Mapped[list["ApiKey"]] = relationship( back_populates="created_by" ) def __repr__(self) -> str: return f"" def has_permission(self, permission: str) -> bool: """Check if user has a specific permission through any role.""" for role in self.roles: if permission in role.permissions or "*" in role.permissions: return True return False def has_access_to(self, tenant_id: UUID) -> bool: """Check if user belongs to the specified tenant.""" return self.tenant_id == tenant_id class Role(Base): __tablename__ = "roles" id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4) tenant_id: Mapped[UUID | None] = mapped_column( ForeignKey("tenants.id", ondelete="CASCADE") ) name: Mapped[str] = mapped_column(String(100), nullable=False) slug: Mapped[str] = mapped_column(String(100), nullable=False) description: Mapped[str | None] permissions: Mapped[list[str]] = mapped_column(JSON, default=list) is_system: Mapped[bool] = mapped_column(Boolean, default=False) created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow) # Relationships tenant: Mapped["Tenant | None"] = relationship() users: Mapped[list["User"]] = relationship( secondary=user_roles, back_populates="roles" ) __table_args__ = ( # Unique constraint on tenant_id + slug {"sqlite_autoincrement": True}, # Placeholder for unique constraint ) def __repr__(self) -> str: return f""