# Kalimera Workspace-User-Pending Request Management
- Created at : 2025-06-14
- Created by : Manoj Vala
## Overview:
This documentation outlines the implementation of a scalable permission system combining role-based access control (RBAC) with a star-topology workspace hierarchy. The system enables:
- Single accounts across multiple workspaces with distinct roles
- Centralized policy management based on workspace type and role
- Flat hierarchy where all workspaces link to a root parent
- Module-level privileges (VIEW/MANAGE) enforced via decorators
- Workspace switching without reauthentication
Designed for Flask/FastAPI with PostgreSQL and Redis, this solution balances flexibility with performance.
---
Let me know if you'd prefer to emphasize different aspects or adjust the technical tone.
---
## Executive Summary
Kalimera is a multi‑workspace SaaS platform where a single account can:
* belong to many workspaces (a.k.a. *tenants*)
* hold a distinct **role** (ADMIN / USER) in each workspace
* gain **module‑level privileges** derived from a central policy table keyed by `(workspace_type, role, module)`
* switch the *current* workspace at any time with no new JWT
Everything is surfaced through **Flask‑RESTful Resources** with the same decorator stack you already use (`@setup_required`, `@login_required`, `@account_initialization_required`) plus one new **RBAC decorator** `@policy_required(module, privilege)`.
---
## 1 Glossary
| Term | Meaning |
| ------------- | - |
| **Account** | A human login (`kalimera_accounts`). |
| **Workspace** | A container for data and members (`kalimera_workspaces`). |
| **Role** | `ADMIN` or `USER` – stored in join table. |
| **Module** | Functional slice of the UI (`campaign_mgmt`, `reports`, `role_mgmt`, …). |
| **Privilege** | `VIEW` or `MANAGE`. |
| **Policy** | Row in `kalimera_workspace_role_policies` that maps `(workspace_type, role, module) → privilege`. |
| **ROOT** | Hard‑coded system approver (`id = 1`). |
---
## 2 Architecture Layers
```
┌───────────────┐ Flask‑RESTful ┌────────────────┐
│ Resource │ decorators │ Service │
│ (HTTP) │───────────────▶ │ (business) │
└───────────────┘ └────────────────┘
▲ ▲
│ SQLAlchemy models │ Redis cache
▼ ▼
┌──────────────────┐ ┌──────────────────┐
│ PostgreSQL │ │ Redis │
│ (kalimera_*) │ │ / JWT │
└──────────────────┘ └──────────────────┘
```
---
## 3 Data Model (DDL Fragment)
```python
class WorkspaceType(enum.Enum):
personal = "personal"
team = "team"
# Add more as needed
class RoleEnum(enum.Enum):
admin = "admin"
member = "member"
viewer = "viewer"
# Add more as needed
class PrivilegeEnum(enum.Enum):
read = "read"
write = "write"
delete = "delete"
# Add more as needed
# Models
class KalimeraAccount(Base):
__tablename__ = "kalimera_accounts"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
email = Column(String, unique=True, nullable=False)
name = Column(String, nullable=False)
password = Column(String)
hard_coded_root = Column(Boolean, default=False)
class KalimeraWorkspace(Base):
__tablename__ = "kalimera_workspaces"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
name = Column(String, nullable=False)
workspace_type = Column(Enum(WorkspaceType), nullable=False)
created_by = Column(UUID(as_uuid=True), ForeignKey("kalimera_accounts.id"))
created_at = Column(DateTime, default=datetime.utcnow)
creator = relationship("Account")
class KalimeraWorkspaceAccountJoin(Base):
__tablename__ = "kalimera_workspace_account_joins"
workspace_id = Column(UUID(as_uuid=True), ForeignKey("kalimera_workspaces.id"))
account_id = Column(UUID(as_uuid=True), ForeignKey("kalimera_accounts.id"))
role = Column(Enum(RoleEnum), nullable=False)
current = Column(Boolean, default=False)
joined_at = Column(DateTime, default=datetime.utcnow)
__table_args__ = (
PrimaryKeyConstraint("workspace_id", "account_id"),
)
class KalimeraModule(Base):
__tablename__ = "kalimera_modules"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
code = Column(String, unique=True, nullable=False)
label = Column(String, nullable=False)
class KalimeraWorkspaceRolePolicy(Base):
__tablename__ = "kalimera_workspace_role_policies"
workspace_type = Column(Enum(WorkspaceType), primary_key=True)
role = Column(Enum(RoleEnum), primary_key=True)
module_id = Column(UUID(as_uuid=True), ForeignKey("kalimera_modules.id"), primary_key=True)
privilege = Column(Enum(PrivilegeEnum), nullable=False)
updated_by = Column(UUID(as_uuid=True), ForeignKey("kalimera_accounts.id"))
module = relationship("Module")
updated_by_account = relationship("Account", foreign_keys=[updated_by])
```
---
## 4 Permission Model
```
(privilege = MANAGE) ⇒ full CRUD
(privilege = VIEW) ⇒ read‑only
Lookup path on every check:
workspace.id ──▶ workspace_type
+ role (from join row)
+ module.code
─────────────────────▶ kalimera_workspace_role_policies.privilege
```
`@policy_required(module_code, Privilege.MANAGE)` wraps this logic; service calls can reuse `PermissionService.can()` directly.
---
## 5 Workspace Architecture:
```mermaid
graph TD
ROOT-->A[Workspace A: parent_id=NULL]
A-->B[Workspace B]-->C[Workspace C]
A-->D[Workspace D]
style A stroke:#ff0000,stroke-width:4px
```
**Root Workspace**: First workspace created (parent_id = NULL)
**Star Children**: All subsequent workspaces point to root
**Data Isolation**: Queries scope to current workspace by default
**Inheritance**: Optional include_parent=True parameter for services
---
## 6 User Flow:
```mermaid
graph TD
A[Account] --> B[Workspace Memberships]
B --> C[Role: ADMIN/USER]
B --> D[Current Workspace]
E[Workspace] --> F[Parent Workspace]
E --> G[Type: personal/team]
C --> H[Policy Engine]
I[Module] --> H
H --> J[Privilege: VIEW/MANAGE]
```
---
## 7 Endpoint Catalogue (v1)
| # | Method & Path | Purpose | Decorators / Gate |
| ----------------- | --------- | ------ | ------------ |
| **Auth** | | | |
| 1 | `POST /auth/register` | Create pending request | *(public)* |
| 2 | `POST /auth/login` | Issue 30‑day JWT | *(public)* |
| **Root‑internal** | | | |
| 3 | `POST /pending-requests/{id}/approve` | Approve signup | `current_user.id == ROOT_ID` |
| 4 | `POST /pending-requests/{id}/reject` | Reject signup | ROOT |
| **Workspaces** | | | |
| 5 | `GET /workspaces` | List workspaces | std decorators |
| 6 | `POST /workspaces` | Create | `@policy_required('workspace_settings', MANAGE)` |
| 7 | `PATCH /workspaces/{id}` | Rename/plan | same gate |
| 8 | `DELETE /workspaces/{id}` | Archive | same gate |
| **Membership** | | | |
| 9 | `POST /workspaces/{id}/invite` | Invite user | `@policy_required('user_mgmt', MANAGE)` |
| 10 | `PATCH /workspaces/{id}/members/{account_id}` | Change role | `@policy_required('role_mgmt', MANAGE)` |
| **Switch** | | | |
| 11 | `POST /workspaces/switch` | Flip current flag | membership check |
| **Policies** | | | |
| 12 | `PATCH /policies/{workspace_type}/{role}` | Edit matrix | `@policy_required('role_mgmt', MANAGE)` |
| **Context** | | | |
| 13 | `GET /account` | Profile + memberships blob | std decorators |
---
## 8 Decorator Stack
```python
@setup_required
@login_required
@account_initialization_required
@policy_required("module_code", Privilege.MANAGE)
def post(...):
...
```
*`policy_required` must run **after** auth decorators so `current_user` is available.*
---
## 9 Service Responsibilities
| Service | Core Methods |
| ----- | -- |
| **PermissionService** | `can(db, account_id, workspace_id, module, need)` |
| **WorkspaceService** | `list_for_account`, `create`, `update`, `delete`, `switch` |
| **MembershipService** | `invite`, `change_role`, `remove` |
| **PolicyService** | `patch_policy` (upsert rows) |
| **PendingRequestService** | `approve`, `reject` |
| **ContextService** | `make_account_blob` (builds `/account` JSON) |
---
## 10 Resource Templates
```python
class WorkspaceApi(Resource):
decorators = [
setup_required, login_required, account_initialization_required
]
@marshal_with(workspace_fields)
def get(self, workspace):
return workspace
@policy_required("workspace_settings", Privilege.MANAGE)
@marshal_with(workspace_fields)
def patch(self, workspace):
patch = request.get_json()
return WorkspaceService().update(workspace, patch)
@policy_required("workspace_settings", Privilege.MANAGE)
def delete(self, workspace):
WorkspaceService().delete(workspace)
return {"result": "success"}, 204
```
Routes:
```python
api.add_resource(WorkspacesApi, "/workspaces")
api.add_resource(WorkspaceApi, "/workspaces/<uuid:workspace>")
api.add_resource(WorkspaceSwitchApi, "/workspaces/switch")
api.add_resource(PolicyApi, "/policies/<string:ws_type>/<string:role>")
api.add_resource(AccountApi, "/account")
```
---
## 11 Account Context Blob (`GET /account`)
```python
def account_blob():
joins = (
db.query(KalimeraWorkspaceAccountJoin, KalimeraWorkspace)
.join(KalimeraWorkspace, KalimeraWorkspace.id == KalimeraWorkspaceAccountJoin.workspace_id)
.filter(KalimeraWorkspaceAccountJoin.account_id == acc.id)
.all())
resp = {"user": {"id": acc.id, "email": acc.email, "name": acc.name},
"memberships": [], "default_workspace_id": None}
for join, ws in joins:
privs = {} # assemble by joining policy+module
resp["memberships"].append({
"workspace_id": ws.id,
"workspace_name": ws.name,
"workspace_type": ws.workspace_type,
"role": join.role,
"module_privileges": privs,
"current": join.current,
})
if join.current:
resp["default_workspace_id"] = ws.id
return resp
```
```jsonc
{
"user": { "id": "392b...", "email": "jane@x", "name": "Jane" },
"memberships": [
{
"workspace_id": "w1",
"workspace_name": "Acme",
"workspace_type": "PLATFORM",
"role": "ADMIN",
"module_privileges": {
"campaign_mgmt": "manage",
"reports": "view",
"role_mgmt": "manage"
},
"current": true
},
...
],
"default_workspace_id": "w1"
}
```
*Built once on login, cached 30 min in Redis.*
---
| # | Method & Path | Auth | Body / Query | Success (200/201) | Permission Gate | |
| ------ | ----- | ----- | --- | ---------- | ----- | -- |
| **Auth**| | || | | |
| 1 | `POST /auth/register` | public | `{ email, name, password }` | `{ pending_request_id }` | — | |
| 2 | `POST /auth/login`| public | `{ email, password }`| `{ access_token, expires_in }` | — | |
| **ROOT‑only** | | | | | | |
| 3 | `POST /pending-requests/{id}/approve` | JWT (root) | `{ workspace_name?, workspace_type?, modules? }` | `204` | root account | |
| 4 | `POST /pending-requests/{id}/reject`| JWT (root) | `{ reason }` | `204` | root account | |
| **Workspaces** | | | | | | |
| 5 | `POST /workspaces`| JWT| `{ name, workspace_type }` | `{ workspace_id }` | caller role = ADMIN | |
| 6 | `GET /workspaces` | JWT| —| `[ ...minimal list... ]` | any | |
| 7 | `PATCH /workspaces/{id}` | JWT| `{ name?, plan?, status? }` | `204` | ADMIN + policy(`workspace_settings`,`MANAGE`) | |
| **Membership** | | | | | | |
| 8 | `POST /workspaces/{id}/invite` | JWT| `{ email, role, modules? }` | `202 Accepted` | policy(`user_mgmt`,`MANAGE`)| |
| 9 | `PATCH /workspaces/{id}/members/{account_id}` | JWT| `{ role }` | `204` | policy(`role_mgmt`,`MANAGE`)| |
| **Switch** | | | | | | |
| 10 | `POST /workspaces/switch` | JWT| `{ workspace_id }` | `{ new_current_id }` | membership exists | |
| **Policies** | | | | | | |
| 11 | `PATCH /policies/{workspace_type}/{role}` | JWT| \`{ module\_privileges: { code: "VIEW| MANAGE" } }\`| `204` | caller must have `role_mgmt=MANAGE` |
| **Context blob** | | | | | | |
| 12 | `GET /account` | JWT| —| see §4 payload | any | |
---
---
## 12 · Service Layer (`services.py`)
```python
from sqlalchemy.orm import Session
from models import (
KalimeraAccount, KalimeraWorkspace, KalimeraWorkspaceAccountJoin,
KalimeraModule, KalimeraWorkspaceRolePolicy, WorkspaceType, Role, Privilege
)
class PermissionService:
@staticmethod
def module_privilege(db: Session, workspace_type, role, module_code):
policy = (db.query(KalimeraWorkspaceRolePolicy)
.join(KalimeraModule, KalimeraModule.id == KalimeraWorkspaceRolePolicy.module_id)
.filter(KalimeraWorkspaceRolePolicy.workspace_type == workspace_type,
KalimeraWorkspaceRolePolicy.role == role,
KalimeraModule.code == module_code)
.first())
return policy.privilege if policy else None
@staticmethod
def can(db, account_id, workspace_id, module_code, need: Privilege):
w = db.query(KalimeraWorkspace).get(workspace_id)
mem = db.query(KalimeraWorkspaceAccountJoin).get((workspace_id, account_id))
if not w or not mem:
return False
level = PermissionService.module_privilege(db, w.workspace_type, mem.role, module_code)
return (level == Privilege.MANAGE) or (level == Privilege.VIEW and need == Privilege.VIEW)
class WorkspaceService:
@staticmethod
def switch(db: Session, account_id: str, workspace_id: str):
# validate membership
join = db.query(KalimeraWorkspaceAccountJoin).get((workspace_id, account_id))
if not join:
raise ValueError("Not a member")
# unset previous
db.query(KalimeraWorkspaceAccountJoin)\
.filter_by(account_id=account_id, current=True)\
.update({KalimeraWorkspaceAccountJoin.current: False})
join.current = True
db.commit()
return workspace_id
```
*(Add more services for invites, policy edit, etc.)*
---
## 13 · FastAPI Router (`api.py`)
```python
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List, Dict
from pydantic import BaseModel, EmailStr
from models import Privilege
from services import PermissionService, WorkspaceService
from deps import get_db, get_current_account, get_current_workspace
router = APIRouter(prefix="/workspaces", tags=["workspaces"])
# ───── Switch Workspace ─────
class SwitchReq(BaseModel):
workspace_id: str
class SwitchResp(BaseModel):
new_current_id: str
@router.post("/switch", response_model=SwitchResp)
def switch_workspace(req: SwitchReq,
db: Session = Depends(get_db),
acc = Depends(get_current_account)):
try:
wid = WorkspaceService.switch(db, acc.id, req.workspace_id)
return {"new_current_id": wid}
except ValueError:
raise HTTPException(status_code=404, detail="Workspace not found")
# ───── Policy Editor ─────
class PolicyPatch(BaseModel):
module_privileges: Dict[str, Privilege] # {"campaign_mgmt":"MANAGE"}
@router.patch("/policies/{ws_type}/{role}", status_code=204)
def edit_policy(ws_type: str, role: str, body: PolicyPatch,
db: Session = Depends(get_db),
acc = Depends(get_current_account),
cur_ws = Depends(get_current_workspace)):
if not PermissionService.can(db, acc.id, cur_ws.id, "role_mgmt", Privilege.MANAGE):
raise HTTPException(status_code=403)
# upsert loop ...
return
```
*`deps.py` supplies `get_current_account` (from JWT) and `get_current_workspace` (joins table `current=True`).*
---
## 14 · `GET /account` Implementation Sketch
```python
@router.get("/account")
def account_blob(db: Session = Depends(get_db),
acc = Depends(get_current_account)):
joins = (db.query(KalimeraWorkspaceAccountJoin, KalimeraWorkspace)
.join(KalimeraWorkspace, KalimeraWorkspace.id == KalimeraWorkspaceAccountJoin.workspace_id)
.filter(KalimeraWorkspaceAccountJoin.account_id == acc.id)
.all())
resp = {"user": {"id": acc.id, "email": acc.email, "name": acc.name},
"memberships": [], "default_workspace_id": None}
for join, ws in joins:
privs = {} # assemble by joining policy+module
resp["memberships"].append({
"workspace_id": ws.id,
"workspace_name": ws.name,
"workspace_type": ws.workspace_type,
"role": join.role,
"module_privileges": privs,
"current": join.current,
})
if join.current:
resp["default_workspace_id"] = ws.id
return resp
```
---
## 15 · Seed Data
```python
seed_modules = [
("campaign_mgmt", "Campaign Management"),
("reports", "Reporting"),
("billing", "Billing"),
("role_mgmt", "Role Management"),
("user_mgmt", "User Management"),
]
# insert into kalimera_modules once.
# Example default policy: PLATFORM.ADMIN gets MANAGE on everything
for mod_code, _ in seed_modules:
db.add(KalimeraWorkspaceRolePolicy(
workspace_type=WorkspaceType.PLATFORM,
role=Role.ADMIN,
module_id=module_id_lookup[mod_code],
privilege=Privilege.MANAGE,
updated_by=root_id
))
```
---
## 16 Caching & Performance
* 🚀 **Zero DB look‑ups per request** if you keep `/account` blob in Redis and JWT in header.
* **Cache bust** triggers: role change, policy edit, workspace join/leave.
> [!IMPORTANT]
> - All indexes of the tables will created based on the use case when code implementation starts.
> - For handling role based action we will create some new decorators to restricts the actions performed by every roles.
---
Sign in with Wallet
Connect another wallet