99 lines
3.5 KiB
Python
99 lines
3.5 KiB
Python
|
from sqlalchemy import ForeignKey, Integer, String, select
|
||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||
|
|
|
||
|
|
from database import Base
|
||
|
|
|
||
|
|
from .associations import role_permissions, user_roles
|
||
|
|
from .enums import Permission
|
||
|
|
|
||
|
|
|
||
|
|
class Role(Base):
|
||
|
|
__tablename__ = "roles"
|
||
|
|
|
||
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||
|
|
name: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
|
||
|
|
description: Mapped[str] = mapped_column(String(255), nullable=True)
|
||
|
|
|
||
|
|
# Relationship to users
|
||
|
|
users: Mapped[list["User"]] = relationship(
|
||
|
|
"User",
|
||
|
|
secondary=user_roles,
|
||
|
|
back_populates="roles",
|
||
|
|
)
|
||
|
|
|
||
|
|
async def get_permissions(self, db: AsyncSession) -> set[Permission]:
|
||
|
|
"""Get all permissions for this role."""
|
||
|
|
query = select(role_permissions.c.permission).where(
|
||
|
|
role_permissions.c.role_id == self.id
|
||
|
|
)
|
||
|
|
result = await db.execute(query)
|
||
|
|
return {row[0] for row in result.fetchall()}
|
||
|
|
|
||
|
|
async def set_permissions(
|
||
|
|
self, db: AsyncSession, permissions: list[Permission]
|
||
|
|
) -> None:
|
||
|
|
"""Set all permissions for this role (replaces existing)."""
|
||
|
|
delete_query = role_permissions.delete().where(
|
||
|
|
role_permissions.c.role_id == self.id
|
||
|
|
)
|
||
|
|
await db.execute(delete_query)
|
||
|
|
for perm in permissions:
|
||
|
|
insert_query = role_permissions.insert().values(
|
||
|
|
role_id=self.id, permission=perm
|
||
|
|
)
|
||
|
|
await db.execute(insert_query)
|
||
|
|
|
||
|
|
|
||
|
|
class User(Base):
|
||
|
|
__tablename__ = "users"
|
||
|
|
|
||
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||
|
|
email: Mapped[str] = mapped_column(
|
||
|
|
String(255), unique=True, nullable=False, index=True
|
||
|
|
)
|
||
|
|
hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
|
||
|
|
|
||
|
|
# Contact details (all optional)
|
||
|
|
contact_email: Mapped[str | None] = mapped_column(String(255), nullable=True)
|
||
|
|
telegram: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||
|
|
signal: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||
|
|
nostr_npub: Mapped[str | None] = mapped_column(String(63), nullable=True)
|
||
|
|
|
||
|
|
# Godfather (who invited this user) - null for seeded/admin users
|
||
|
|
godfather_id: Mapped[int | None] = mapped_column(
|
||
|
|
Integer, ForeignKey("users.id"), nullable=True
|
||
|
|
)
|
||
|
|
godfather: Mapped["User | None"] = relationship(
|
||
|
|
"User",
|
||
|
|
remote_side="User.id",
|
||
|
|
foreign_keys=[godfather_id],
|
||
|
|
)
|
||
|
|
|
||
|
|
# Relationship to roles
|
||
|
|
roles: Mapped[list[Role]] = relationship(
|
||
|
|
"Role",
|
||
|
|
secondary=user_roles,
|
||
|
|
back_populates="users",
|
||
|
|
lazy="selectin",
|
||
|
|
)
|
||
|
|
|
||
|
|
async def get_permissions(self, db: AsyncSession) -> set[Permission]:
|
||
|
|
"""Get all permissions from all roles in a single query."""
|
||
|
|
result = await db.execute(
|
||
|
|
select(role_permissions.c.permission)
|
||
|
|
.join(user_roles, role_permissions.c.role_id == user_roles.c.role_id)
|
||
|
|
.where(user_roles.c.user_id == self.id)
|
||
|
|
)
|
||
|
|
return {row[0] for row in result.fetchall()}
|
||
|
|
|
||
|
|
async def has_permission(self, db: AsyncSession, permission: Permission) -> bool:
|
||
|
|
"""Check if user has a specific permission through any of their roles."""
|
||
|
|
permissions = await self.get_permissions(db)
|
||
|
|
return permission in permissions
|
||
|
|
|
||
|
|
@property
|
||
|
|
def role_names(self) -> list[str]:
|
||
|
|
"""Get list of role names for API responses."""
|
||
|
|
return [role.name for role in self.roles]
|