Add backend role-based access control

Author: shopipoint

backend/app/alembic/versions/b3a8e8a9d5f2_add_user_role.py

@@ -0,0 +1,28 @@
+"""Add user role
+
+Revision ID: b3a8e8a9d5f2
+Revises: fe56fa70289e
+Create Date: 2026-05-05 19:15:00.000000
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = "b3a8e8a9d5f2"
+down_revision = "fe56fa70289e"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "user",
+        sa.Column("role", sa.String(length=20), nullable=False, server_default="member"),
+    )
+    op.execute("UPDATE \"user\" SET role = 'admin' WHERE is_superuser = true")
+
+
+def downgrade():
+    op.drop_column("user", "role")

backend/app/api/deps.py

@@ -11,7 +11,7 @@
 from app.core import security
 from app.core.config import settings
 from app.core.db import engine
-from app.models import TokenPayload, User
+from app.models import TokenPayload, User, UserRole
 
 reusable_oauth2 = OAuth2PasswordBearer(
     tokenUrl=f"{settings.API_V1_STR}/login/access-token"
@@ -49,8 +49,26 @@ def get_current_user(session: SessionDep, token: TokenDep) -> User:
 CurrentUser = Annotated[User, Depends(get_current_user)]
 
 
+def current_role(user: User) -> UserRole:
+    if user.is_superuser:
+        return UserRole.admin
+    return UserRole(user.role)
+
+
+def require_roles(*allowed_roles: UserRole):
+    def role_dependency(current_user: CurrentUser) -> User:
+        if current_role(current_user) not in allowed_roles:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="The user doesn't have enough privileges",
+            )
+        return current_user
+
+    return role_dependency
+
+
 def get_current_active_superuser(current_user: CurrentUser) -> User:
-    if not current_user.is_superuser:
+    if current_role(current_user) != UserRole.admin:
         raise HTTPException(
             status_code=403, detail="The user doesn't have enough privileges"
         )

backend/app/api/main.py

@@ -1,13 +1,14 @@
 from fastapi import APIRouter
 
-from app.api.routes import items, login, private, users, utils
+from app.api.routes import items, login, metrics, private, users, utils
 from app.core.config import settings
 
 api_router = APIRouter()
 api_router.include_router(login.router)
 api_router.include_router(users.router)
 api_router.include_router(utils.router)
 api_router.include_router(items.router)
+api_router.include_router(metrics.router)
 
 
 if settings.ENVIRONMENT == "local":

backend/app/api/routes/metrics.py

@@ -0,0 +1,23 @@
+from typing import Any
+
+from fastapi import APIRouter, Depends
+from sqlmodel import func, select
+
+from app.api.deps import SessionDep, require_roles
+from app.models import Item, User, UserRole
+
+router = APIRouter(prefix="/metrics", tags=["metrics"])
+
+
+@router.get("/", dependencies=[Depends(require_roles(UserRole.admin, UserRole.manager))])
+def read_metrics(session: SessionDep) -> dict[str, Any]:
+    """
+    Return simple operational metrics for privileged users.
+    """
+    user_count = session.exec(select(func.count()).select_from(User)).one()
+    item_count = session.exec(select(func.count()).select_from(Item)).one()
+    return {
+        "users": user_count,
+        "items": item_count,
+        "summary": "Demo insights visible to admins and managers.",
+    }

backend/app/api/routes/users.py

@@ -8,7 +8,9 @@
 from app.api.deps import (
     CurrentUser,
     SessionDep,
+    current_role,
     get_current_active_superuser,
+    require_roles,
 )
 from app.core.config import settings
 from app.core.security import get_password_hash, verify_password
@@ -20,6 +22,7 @@
     UserCreate,
     UserPublic,
     UserRegister,
+    UserRole,
     UsersPublic,
     UserUpdate,
     UserUpdateMe,
@@ -31,7 +34,7 @@
 
 @router.get(
     "/",
-    dependencies=[Depends(get_current_active_superuser)],
+    dependencies=[Depends(require_roles(UserRole.admin, UserRole.manager))],
     response_model=UsersPublic,
 )
 def read_users(session: SessionDep, skip: int = 0, limit: int = 100) -> Any:
@@ -169,7 +172,7 @@ def read_user_by_id(
     user = session.get(User, user_id)
     if user == current_user:
         return user
-    if not current_user.is_superuser:
+    if current_role(current_user) not in {UserRole.admin, UserRole.manager}:
         raise HTTPException(
             status_code=403,
             detail="The user doesn't have enough privileges",

backend/app/core/db.py

@@ -2,7 +2,7 @@
 
 from app import crud
 from app.core.config import settings
-from app.models import User, UserCreate
+from app.models import User, UserCreate, UserRole
 
 engine = create_engine(str(settings.SQLALCHEMY_DATABASE_URI))
 
@@ -29,5 +29,26 @@ def init_db(session: Session) -> None:
             email=settings.FIRST_SUPERUSER,
             password=settings.FIRST_SUPERUSER_PASSWORD,
             is_superuser=True,
+            role=UserRole.admin,
         )
         user = crud.create_user(session=session, user_create=user_in)
+    elif user.role != UserRole.admin:
+        user.role = UserRole.admin
+        user.is_superuser = True
+        session.add(user)
+        session.commit()
+
+    seed_users = [
+        ("manager@example.com", "changethis", UserRole.manager, "Demo Manager"),
+        ("member@example.com", "changethis", UserRole.member, "Demo Member"),
+    ]
+    for email, password, role, full_name in seed_users:
+        existing_user = session.exec(select(User).where(User.email == email)).first()
+        if not existing_user:
+            user_in = UserCreate(
+                email=email,
+                password=password,
+                role=role,
+                full_name=full_name,
+            )
+            crud.create_user(session=session, user_create=user_in)

backend/app/crud.py

@@ -4,12 +4,18 @@
 from sqlmodel import Session, select
 
 from app.core.security import get_password_hash, verify_password
-from app.models import Item, ItemCreate, User, UserCreate, UserUpdate
+from app.models import Item, ItemCreate, User, UserCreate, UserRole, UserUpdate
 
 
 def create_user(*, session: Session, user_create: UserCreate) -> User:
+    role = UserRole.admin if user_create.is_superuser else user_create.role
     db_obj = User.model_validate(
-        user_create, update={"hashed_password": get_password_hash(user_create.password)}
+        user_create,
+        update={
+            "hashed_password": get_password_hash(user_create.password),
+            "role": role,
+            "is_superuser": role == UserRole.admin,
+        },
     )
     session.add(db_obj)
     session.commit()
@@ -24,6 +30,10 @@ def update_user(*, session: Session, db_user: User, user_in: UserUpdate) -> Any:
         password = user_data["password"]
         hashed_password = get_password_hash(password)
         extra_data["hashed_password"] = hashed_password
+    if user_data.get("role") == UserRole.admin:
+        extra_data["is_superuser"] = True
+    elif user_data.get("role") in {UserRole.manager, UserRole.member}:
+        extra_data["is_superuser"] = False
     db_user.sqlmodel_update(user_data, update=extra_data)
     session.add(db_user)
     session.commit()

backend/app/models.py

@@ -1,20 +1,33 @@
 import uuid
 from datetime import datetime, timezone
+from enum import Enum
 
 from pydantic import EmailStr
-from sqlalchemy import DateTime
+from sqlalchemy import Column, DateTime, String
 from sqlmodel import Field, Relationship, SQLModel
 
 
 def get_datetime_utc() -> datetime:
     return datetime.now(timezone.utc)
 
 
+class UserRole(str, Enum):
+    admin = "admin"
+    manager = "manager"
+    member = "member"
+
+
 # Shared properties
 class UserBase(SQLModel):
     email: EmailStr = Field(unique=True, index=True, max_length=255)
     is_active: bool = True
     is_superuser: bool = False
+    role: UserRole = Field(
+        default=UserRole.member,
+        sa_column=Column(
+            String(20), nullable=False, server_default=UserRole.member.value
+        ),
+    )
     full_name: str | None = Field(default=None, max_length=255)
 
 

backend/tests/api/routes/test_metrics.py

@@ -0,0 +1,28 @@
+from fastapi.testclient import TestClient
+from sqlmodel import Session
+
+from app.core.config import settings
+from app.models import UserRole
+from tests.utils.user import create_user_with_role, user_authentication_headers
+
+
+def test_manager_can_read_metrics(client: TestClient, db: Session) -> None:
+    manager, password = create_user_with_role(db=db, role=UserRole.manager)
+    headers = user_authentication_headers(
+        client=client, email=manager.email, password=password
+    )
+
+    r = client.get(f"{settings.API_V1_STR}/metrics/", headers=headers)
+
+    assert r.status_code == 200
+    assert "users" in r.json()
+    assert "items" in r.json()
+
+
+def test_member_cannot_read_metrics(
+    client: TestClient, normal_user_token_headers: dict[str, str]
+) -> None:
+    r = client.get(f"{settings.API_V1_STR}/metrics/", headers=normal_user_token_headers)
+
+    assert r.status_code == 403
+    assert r.json()["detail"] == "The user doesn't have enough privileges"

backend/tests/api/routes/test_users.py

@@ -7,8 +7,12 @@
 from app import crud
 from app.core.config import settings
 from app.core.security import verify_password
-from app.models import User, UserCreate
-from tests.utils.user import create_random_user
+from app.models import User, UserCreate, UserRole
+from tests.utils.user import (
+    create_random_user,
+    create_user_with_role,
+    user_authentication_headers,
+)
 from tests.utils.utils import random_email, random_lower_string
 
 
@@ -519,3 +523,36 @@ def test_delete_user_without_privileges(
     )
     assert r.status_code == 403
     assert r.json()["detail"] == "The user doesn't have enough privileges"
+
+
+def test_manager_can_retrieve_users(client: TestClient, db: Session) -> None:
+    manager, password = create_user_with_role(db=db, role=UserRole.manager)
+    headers = user_authentication_headers(
+        client=client, email=manager.email, password=password
+    )
+
+    r = client.get(f"{settings.API_V1_STR}/users/", headers=headers)
+
+    assert r.status_code == 200
+    assert "data" in r.json()
+
+
+def test_member_cannot_retrieve_users(
+    client: TestClient, normal_user_token_headers: dict[str, str]
+) -> None:
+    r = client.get(f"{settings.API_V1_STR}/users/", headers=normal_user_token_headers)
+
+    assert r.status_code == 403
+    assert r.json()["detail"] == "The user doesn't have enough privileges"
+
+
+def test_manager_cannot_create_user(client: TestClient, db: Session) -> None:
+    manager, password = create_user_with_role(db=db, role=UserRole.manager)
+    headers = user_authentication_headers(
+        client=client, email=manager.email, password=password
+    )
+    data = {"email": random_email(), "password": random_lower_string()}
+
+    r = client.post(f"{settings.API_V1_STR}/users/", headers=headers, json=data)
+
+    assert r.status_code == 403