lilJ/spidy-look-at-it
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

some updated

Browse Source
This commit is contained in:
Kulvir Singh
2025-12-26 22:39:54 +05:30
parent b22477ef13
commit b92b43c4bc
2 changed files with 231 additions and 565 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
01-schema-refactoring.md
View File

@@ -9,11 +9,6 @@
- to prevent such scenarios, funny strategies need to be implemented in code
- **conclusion** unnecessary complexity is introduced.
### **Funny ways versioning**
- versioning means having older versions and newer versions and maintains a relation b/w them tooo so we know what change and how.
- currently no such versioning system is there.
### **complex queries for simple things**
```python
@@ -70,10 +65,7 @@ class Document():
document_type = models.CharField(
max_length=20,
choices=DocumentType.choices
)
name = models.CharField(max_length=500)
blob_url = models.URLField(max_length=2000)
file_type = models.CharField(max_length=20, choices=FileType.choices)
) name = models.CharField(max_length=500) blob_url = models.URLField(max_length=2000) file_type = models.CharField(max_length=20, choices=FileType.choices)
size_bytes = models.BigIntegerField()
total_pages = models.IntegerField(null=True, blank=True)
@@ -129,24 +121,12 @@ class Document():
```python
class Checklist():
document = models.ForeignKey(
"Document",
on_delete=models.SET_NULL,
null=True,
blank=True,
)
id = models.UUIDField()
id = models.UUIDField(db_index=True)
name = models.CharField(max_length=200)
description = models.TextField(blank=True)
version = models.IntegerField(default=1)
previous_version = models.ForeignKey(
"self",
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="next_versions"
)
is_current = models.BooleanField(default=True)
document = models.ForeignKey("Document", on_delete=models.SET_NULL)
organization = models.ForeignKey(
"Organization",
@@ -159,8 +139,6 @@ class Checklist():
null=True,
)
name = models.CharField(max_length=200)
description = models.TextField(blank=True)
```
### Keyterm
@@ -174,33 +152,21 @@ class Keyterm():
id = models.UUIDField(db_index=True)
version = models.IntegerField(default=1)
previous_version = models.ForeignKey(
"self",
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="next_versions"
)
is_current = models.BooleanField(default=True)
key = models.CharField(max_length=500)
question = models.TextField()
instructions = models.TextField(blank=True)
organization = models.ForeignKey(
"Organization",
on_delete=models.CASCADE,
)
key = models.CharField(max_length=500)
question = models.TextField()
instructions = models.TextField(blank=True)
type = models.CharField(
choices=KeytermType.choices,
default=KeytermType.ANALYSIS
)
expected_answer = models.TextField(null=True, blank=True)
is_active = models.BooleanField(default=True)
```
### Contract
@@ -243,9 +209,6 @@ class Contract():
null=True,
related_name="analyzed_contracts"
)
def get_version_history(self):
return Document.get_version_history(self.document.original_id)
```
### **Analysis MODEL**
@@ -293,7 +256,22 @@ class MSAContract():
auto_renewal = models.BooleanField(default=False)
```
### **Strategy Pattern** for type specific logic
**Benefits**:
- `Contract` table stays clean
- type specific data is normalized
- no changes to existing code when adding new types, adding new contract type = new detail table + new handler
- lastly, easy to test handlers in isolation
**Concern**: Separate tables per contract type could lead to "schema explosion"
BUT....
- Current system has 10-15 distinct contract **categories**
- Many categories share similar metadata structures
- Most contract types might only need generic metadata storage
---
### **Strategy Pattern** for structuring code.
```python
class IContractHandler():
@@ -343,14 +321,6 @@ class ContractService:
return contract
```
**Benefits**:
- `Contract` table stays clean
- type specific data is normalized
- no changes to existing code when adding new types, adding new contract type = new detail table + new handler
- lastly, easy to test handlers in isolation
---
### Dependency Injection
```python
@@ -444,8 +414,6 @@ class ServiceFactory:
)
```
---
### usage in views or anything
```python
@@ -464,3 +432,4 @@ def upload_contract(request):
return Response({"contract_id": str(contract.id)})
```

675
02-multiple-dashboards.md
View File

@@ -6,201 +6,124 @@
- separate email agents
- user can belong to multiple teams
---
## RBAC
### Permissions
**Key Concepts**:
1. **Permission** - Individual action (e.g., `contract:create`, `checklist:edit`)
2. **Role** - Collection of permissions (ROOT, ADMIN, VIEWER)
3. **TeamMembership** - Links user to team with a role
- users belong to an **Organization**
- users can be granted access to multiple **Teams**
- each team access has a specific **Role**
- roles contain **Permissions**
- all permissions are stored in `Permission` table
- roles stored in `Role` table
- roles has many-to-many relationship with `Permission` table
- link users to teams with specific roles
- users can have different roles in different teams
- permissions are derived from the assigned role
** [Entity Relationship Schema](https://app.eraser.io/workspace/ZSPoTCtYZRIwToc57lyO?origin=share)**
### **Database Schema**
```python
class Permission(models.TextChoices):
CONTRACT_VIEW = "contract:view"
CONTRACT_CREATE = "contract:create"
CONTRACT_ANALYZE = "contract:analyze"
CONTRACT_DELETE = "contract:delete"
CHECKLIST_VIEW = "checklist:view"
CHECKLIST_CREATE = "checklist:create"
CHECKLIST_EDIT = "checklist:edit"
CHECKLIST_DELETE = "checklist:delete"
CHECKLIST_SET_ACTIVE = "checklist:set_active"
TEAM_VIEW = "team:view"
TEAM_VIEW_ALL = "team:view:all" # super admin - see all teams
TEAM_MANAGE_MEMBERS = "team:manage_members"
TEAM_EDIT_SETTINGS = "team:edit_settings"
TEAM_DELETE = "team:delete"
EMAIL_AGENT_VIEW = "email_agent:view"
EMAIL_AGENT_CONFIGURE = "email_agent:configure"
USER_INVITE = "user:invite"
USER_MANAGE = "user:manage"
class Organization():
name = models.CharField(max_length=200)
email = models.EmailField()
```
```python
class User():
id = models.UUIDField()
name = models.CharField(max_length=100)
email = models.EmailField(max_length=100)
organization = models.ForeignKey(
Organization,
on_delete=models.CASCADE,
db_column="organization_id",
)
is_superuser = models.BooleanField(default=False) # root admin
```
**Permissions**: permissions
```python
class Permission():
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
code = models.CharField(max_length=100, unique=True) # e.g., "contract:create"
name = models.CharField(max_length=200) # e.g., "Create Contract"
category = models.CharField(max_length=50) # e.g., "contract", "team", "checklist"
# description = models.TextField(blank=True)
```
**Role**: Collection of permissions
```python
class Role():
class RoleType(models.TextChoices):
ADMIN = "ADMIN"
MASTER = "MASTER"
VIEWER = "VIEWER"
name = models.CharField(max_length=50, unique=True)
name = models.CharField(max_length=50) # e.g., "ROOT", "ADMIN", "VIEWER"
description = models.TextField(blank=True)
permissions = models.JSONField(default=list)
is_system = models.BooleanField(default=False)
ROLE_PERMISSIONS = {
"ADMIN": [
"contract:view", "contract:create", "contract:analyze", "contract:delete",
"checklist:view", "checklist:create", "checklist:edit", "checklist:delete", "checklist:set_active",
"team:view", "team:view:all", "team:manage_members", "team:edit_settings", "team:delete",
"email_agent:view", "email_agent:configure",
"user:invite", "user:manage",
],
"MASTER": [
"contract:view", "contract:create", "contract:analyze",
"checklist:view", "checklist:create", "checklist:edit", "checklist:set_active",
"team:view", "team:manage_members",
"email_agent:view",
],
"VIEWER": [
"contract:view",
"checklist:view",
"team:view",
],
}
is_system = models.BooleanField(default=False) # Flag for default/system roles
organization = models.ForeignKey(
Organization,
on_delete=models.CASCADE,
related_name="roles"
)
permissions = models.ManyToManyField(
Permission,
related_name="roles",
blank=True
)
```
### Permission
- each organization has its own set of roles (along with default 3 `system` roles i.e. "ROOT", "ADMIN", "VIEWER")
- `system_roles` created automatically with standard permissions, but can be customized per organization
```python
def get_user_permissions(user: User, team_id: str = None) -> set:
membership = TeamMembership.objects.filter(
user=user,
team_id=team_id,
deleted_at__isnull=True
).select_related("role").first()
if not membership:
return set()
return set(membership.role.permissions)
def has_permission(user: User, permission: str, team_id: str = None) -> bool:
return permission in get_user_permissions(user, team_id)
def has_team_access(user: User, team_id: str) -> bool:
if has_permission(user, "team:view:all"):
return True
return TeamMembership.objects.filter(
user=user,
team_id=team_id,
deleted_at__isnull=True
).exists()
```
```python
def require_permission(permission: str, team_id: UUID)
def decorator(view_func):
@wraps(view_func)
def wrapper(request, *args, **kwargs):
user = request.user
team_id = kwargs.get(team_id_param) or request.data.get(team_id_param)
if not has_permission(user, permission, team_id):
return Response(
{ "error": "Permission denied" },
status=status.HTTP_403_FORBIDDEN
)
return view_func(request, *args, **kwargs)
return wrapper
return decorator
def require_team_access(team_id_param: str = "team_id"):
"""Decorator to check team membership"""
def decorator(view_func):
@wraps(view_func)
def wrapper(request, *args, **kwargs):
team_id = kwargs.get(team_id_param)
if not has_team_access(request.user, team_id):
return Response(
{"error": "Team not found"},
status=status.HTTP_404_NOT_FOUND
)
return view_func(request, *args, **kwargs)
return wrapper
return decorator
```
### Usage in Views
```python
@api_view(["POST"])
@permission_classes([IsAuthenticated])
@require_team_access("team_id")
@require_permission("contract:create", "team_id")
def upload_contract(request, team_id):
# user has access to team AND has contract:create permission
```
---
## Team
**Team**: workspace within an organization
```python
class Team():
name = models.CharField(max_length=200)
description = models.TextField(blank=True)
organization = models.ForeignKey(
"Organization",
Organization,
on_delete=models.CASCADE,
related_name="teams"
)
is_default = models.BooleanField(default=False)
settings = models.JSONField(default=dict)
```
**TeamMembership**: user role assignment in a team
```python
class TeamMembership():
user = models.ForeignKey(
"User",
User,
on_delete=models.CASCADE,
related_name="team_memberships"
)
team = models.ForeignKey(
Team,
on_delete=models.CASCADE,
related_name="memberships"
)
role = models.ForeignKey(
"Role",
on_delete=models.PROTECT,
Role,
on_delete=models.PROTECT, # Prevent deletion of role if in use
related_name="team_memberships"
)
added_by = models.ForeignKey(
"User",
User,
on_delete=models.SET_NULL,
null=True,
related_name="memberships_added"
)
```
```python
class TeamEmailAgent(BaseModel):
class TeamEmailAgent():
team = models.OneToOneField(
Team,
on_delete=models.CASCADE,
@@ -243,370 +166,144 @@ class TeamEmailAgent(BaseModel):
)
```
```python
class Contract():
# existing fields
**Permission Naming Convention**: `resource:action`
- `contract:view`, `contract:create`, `contract:edit`, `contract:delete`, `contract:analyze`
- `team:view`, `team:create`, `team:edit`, `team:delete`, `team:manage_members`
- `checklist:view`, `checklist:create`, `checklist:edit`, `checklist:delete`
- `email_agent:view`, `email_agent:configure`, `email_agent:enable`, `email_agent:disable`
team = models.ForeignKey(
"Team",
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="contracts"
)
```
## permission verifications
```python
class Checklist(BaseModel):
# existing fields
def is_superuser(user: User) -> bool:
return getattr(user, 'is_superuser', False)
# NULL might mean this checklist is available org wide....
team = models.ForeignKey(
"Team",
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="checklists"
)
def get_user_permissions(
user: User,
team_id: Optional[str] = None,
) -> Set[str]:
"""
get all permissions for a user in a team.
"""
cache_key = f"user_perms:{user.id}:{team_id}"
cached_perms = cache.get(cache_key)
if cached_perms is not None:
return set(cached_perms)
membership = TeamMembership.objects.filter(
user=user,
team_id=team_id,
).select_related('role').prefetch_related('role__permissions').first()
if not membership:
return set()
role_permissions = { p.code for p in membership.role.permissions.all() }
if use_cache:
cache.set(cache_key, list(role_permissions))
return role_permissions
def has_permission(
user: User,
permission_code: str,
team_id: Optional[str] = None,
) -> bool:
"""
check if user has a specific permission in a team.
"""
if is_superuser(user):
return True
if not team_id:
raise ValueError("team_id is required for permission check")
permissions = get_user_permissions(user, team_id, use_cache)
if permission_code in permissions:
return True
# NOTE: optional if we are allowing wildcard permissions
# resource, action = permission_code.split(":", 1) if ":" in permission_code else (permission_code, "")
# wildcard_permission = f"{resource}:*"
# if wildcard_permission in permissions:
# return True
return False
```
## Team Management
## **Usage**
```python
class TeamService:
def __init__(self, user: User):
self.user = user
@api_view(["GET"])
@require_permission("contract:view", "team_id")
def list_contracts(request, team_id):
contracts = Contract.objects.filter(team_id=team_id, deleted_at__isnull=True)
return Response({"contracts": list(contracts.values())})
def _check_permission(self, permission: str, team_id: str = None):
if not has_permission(self.user, permission, team_id):
raise PermissionError(f"Missing permission: {permission}")
@transaction.atomic
def create_team(
self,
organization_id: str,
name: str,
description: str = "",
settings: Dict = None,
) -> Team:
# only admins can create teams
self._check_permission("team:view:all")
@api_view(["POST"])
@require_permission("contract:create", "team_id")
def upload_contract(request, team_id):
return Response({"message": "Contract uploaded"})
team = Team.objects.create(
organization_id=organization_id,
name=name,
description=description,
settings=settings or {}
)
# creator becomes admin of this team
admin_role = Role.objects.get(name="ADMIN")
TeamMembership.objects.create(
user=self.user,
team=team,
role=admin_role,
added_by=self.user
)
return team
def get_user_teams(self, user: User) -> List[Dict[str, Any]]:
"""Get all teams a user has access to"""
memberships = TeamMembership.objects.filter(
user=user,
deleted_at__isnull=True
).select_related("team", "role").order_by("team__settings__order", "team__name")
teams = []
for membership in memberships:
team = membership.team
teams.append({
"id": str(team.id),
"name": team.name,
"description": team.description,
"settings": team.settings,
"role": membership.role.name,
"is_default": team.is_default,
})
return teams
@transaction.atomic
def add_member(
self,
team_id: str,
user_id: str,
role_name: str,
) -> TeamMembership:
# requires team:manage_members permission
self._check_permission("team:manage_members", team_id)
team = Team.objects.get(id=team_id)
user = User.objects.get(RowKey=user_id, organization_id=team.organization_id)
role = Role.objects.get(name=role_name.upper())
# prevent assigning higher role than own
my_membership = TeamMembership.objects.get(user=self.user, team=team)
if self._role_level(role) > self._role_level(my_membership.role):
raise PermissionError("Cannot assign role higher than your own")
membership, created = TeamMembership.objects.update_or_create(
team=team,
user=user,
defaults={
"role": role,
"added_by": self.user,
"deleted_at": None,
}
)
return membership
def _role_level(self, role: Role) -> int:
levels = {"VIEWER": 1, "MASTER": 2, "ADMIN": 3}
return levels.get(role.name, 0)
def remove_member(self, team_id: str, user_id: str) -> bool:
self._check_permission("team:manage_members", team_id)
membership = TeamMembership.objects.filter(
team_id=team_id,
user_id=user_id,
deleted_at__isnull=True
).first()
if membership:
membership.deleted_at = timezone.now()
membership.save()
return True
return False
def get_team_stats(self, team_id: str) -> Dict[str, int]:
from assistantService.models import Contract
today = date.today()
next_30 = today + timedelta(days=30)
contracts = Contract.objects.filter(
team_id=team_id,
deleted_at__isnull=True
)
return {
"total_contracts": contracts.count(),
"pending": contracts.filter(analysis_status="PENDING").count(),
"accepted": contracts.filter(analysis_status="ACCEPTED").count(),
"rejected": contracts.filter(analysis_status="REJECTED").count(),
"expiring_in_30_days": contracts.filter(
expiry_date__range=(today, next_30)
).count(),
"expired": contracts.filter(expiry_date__lt=today).count(),
"members": TeamMembership.objects.filter(
team_id=team_id,
deleted_at__isnull=True
).count(),
}
@api_view(["POST"])
@require_superuser()
def create_organization(request):
# Only superusers can create organizations
...
```
## Dashboard Tabs
## API Endpoints
```python
class DashboardService:
def __init__(self, user: User):
self.user = user
### **Dashboard**
def get_dashboard(
self,
team_id: str,
page: int = 1,
page_size: int = 10,
filters: Optional[Dict] = None
) -> Dict[str, Any]:
# uses has_team_access from RBAC
if not has_team_access(self.user, team_id):
raise PermissionError("You don't have access to this team")
- `GET /api/teams/{team_id}/dashboard` - Get team dashboard with contracts and statistics
- **Required Permissions:** Team access + `contract:view`
- **Query Parameters:** `page` (optional), `page_size` (optional), `filters` (optional)
team = Team.objects.get(id=team_id)
- `GET /api/teams/{team_id}/dashboard/stats` - Get team dashboard statistics
- **Required Permissions:** Team access + `contract:view`
query = Contract.objects.filter(
team_id=team_id,
deleted_at__isnull=True
)
### **Checklists**
if filters:
query = self._apply_filters(query, filters)
- `GET /api/teams/{team_id}/checklists` - List all checklists for a team
- **Required Permissions:** Team access + `checklist:view`
query = query.order_by("-created_at")
### **team management within org**
query = query.select_related(
"current_version",
"active_checklist",
"uploaded_by"
)
- `POST /api/teams` - Create a new team in organization
- **Required Permission:** `superuser`
- **Request Body:** `{ "organization_id": "uuid", "name": "string", "description": "string", "settings": {} }`
paginator = Paginator(query, page_size)
page_obj = paginator.get_page(page)
- `PUT /api/teams/{team_id}` - Update team settings
- **Required Permissions:** Team access + `team:edit`
- **Request Body:** `{ "name": "string", "description": "string", "settings": {} }`
contracts = [
self._format_contract(c, team.organization.settings.get("timezone", "UTC"))
for c in page_obj
]
- `DELETE /api/teams/{team_id}` - Delete a team (soft delete)
- **Required Permissions:** Team access + `team:delete`
return {
"team": {
"id": str(team.id),
"name": team.name,
"settings": team.settings,
},
"contracts": contracts,
"pagination": {
"current_page": page,
"total_pages": paginator.num_pages,
"total_count": paginator.count,
"has_next": page_obj.has_next(),
"has_previous": page_obj.has_previous(),
},
"filters": self._get_filter_options(team_id),
}
```
- `GET /api/teams/{team_id}/members` - List all team members with their roles
- **Required Permissions:** Team access + `team:view`
### Email Agent Configuration
- `POST /api/teams/{team_id}/members` - Add a user to a team with a specific role
- **Required Permissions:** Team access + `team:manage_members`
- **Request Body:** `{ "user_id": "uuid", "role_name": "ROOT" | "ADMIN" | "VIEWER" }`
```python
class TeamEmailAgentService:
def __init__(self, user: User):
self.user = user
- `DELETE /api/teams/{team_id}/members/{user_id}` - Remove a user from a team
- **Required Permissions:** Team access + `team:manage_members`
def get_config(self, team_id: str) -> Dict[str, Any]:
if not has_permission(self.user, "email_agent:view", team_id):
raise PermissionError("Cannot view email agent config")
- `PUT /api/teams/{team_id}/members/{user_id}/role` - Update a team member's role
- **Required Permissions:** Team access + `team:manage_members`
- **Request Body:** `{ "role_name": "ROOT" | "ADMIN" | "VIEWER" }`
team = Team.objects.get(id=team_id)
try:
agent = team.email_agent
return {
"is_configured": True,
"inbox_email": agent.inbox_email,
"provider_type": agent.provider_type,
"is_enabled": agent.is_enabled,
"processing_settings": agent.processing_settings,
"subscription_status": {
"active": agent.subscription_id is not None,
"expires_at": agent.subscription_expiry.isoformat() if agent.subscription_expiry else None,
},
"last_sync_at": agent.last_sync_at.isoformat() if agent.last_sync_at else None,
}
except TeamEmailAgent.DoesNotExist:
return {
"is_configured": False,
}
### **Email Agent**
@transaction.atomic
def configure(
self,
team_id: str,
inbox_email: str,
ms_client_id: str,
ms_client_secret: str,
ms_tenant_id: str,
processing_settings: Dict = None
) -> TeamEmailAgent:
# only ADMIN can configure email agent
if not has_permission(self.user, "email_agent:configure", team_id):
raise PermissionError("Cannot configure email agent")
team = Team.objects.get(id=team_id)
graph = GraphEmailService()
try:
token = graph.generate_access_token(
client_id=ms_client_id,
client_secret=ms_client_secret,
tenant_id=ms_tenant_id
)
except Exception as e:
raise ValueError(f"Invalid Microsoft Graph credentials: {e}")
agent, created = TeamEmailAgent.objects.update_or_create(
team=team,
defaults={
"inbox_email": inbox_email,
"ms_client_id": ms_client_id,
"ms_client_secret": ms_client_secret, # TODO: encrypt
"ms_tenant_id": ms_tenant_id,
"processing_settings": processing_settings or {},
}
)
return agent
def enable(self, team_id: str) -> TeamEmailAgent:
if not has_permission(self.user, "email_agent:configure", team_id):
raise PermissionError("Cannot enable email agent")
agent = TeamEmailAgent.objects.get(team_id=team_id)
if not agent.ms_client_id:
raise ValueError("Email agent not configured")
subscription = self._create_subscription(agent)
agent.is_enabled = True
agent.subscription_id = subscription["id"]
agent.subscription_expiry = subscription["expirationDateTime"]
agent.save()
return agent
def disable(self, team_id: str) -> TeamEmailAgent:
if not has_permission(self.user, "email_agent:configure", team_id):
raise PermissionError("Cannot disable email agent")
agent = TeamEmailAgent.objects.get(team_id=team_id)
if agent.subscription_id:
self._delete_subscription(agent)
agent.is_enabled = False
agent.subscription_id = None
agent.subscription_expiry = None
agent.save()
return agent
```
---
## APIs
```python
urlpatterns = [
# team management
path("", views.list_teams), # team:view
path("", views.create_team), # team:view:all (admin only)
path("<uuid:team_id>/", views.get_team), # team:view + membership
path("<uuid:team_id>/", views.update_team), # team:edit_settings
path("<uuid:team_id>/", views.delete_team), # team:delete
# members
path("<uuid:team_id>/members/", views.list_members), # team:view
path("<uuid:team_id>/members/", views.add_member), # team:manage_members
path("<uuid:team_id>/members/<uuid:user_id>/", views.remove_member), # team:manage_members
path("<uuid:team_id>/members/<uuid:user_id>/role/", views.update_member_role),# team:manage_members
# dashboard
path("<uuid:team_id>/dashboard/", views.get_dashboard), # contract:view
path("<uuid:team_id>/dashboard/stats/", views.get_dashboard_stats),
# checklists
path("<uuid:team_id>/checklists/", views.list_team_checklists), # checklist:view
# email agent
path("<uuid:team_id>/email-agent/", views.get_email_agent_config), # email_agent:view
path("<uuid:team_id>/email-agent/", views.configure_email_agent), # email_agent:configure
path("<uuid:team_id>/email-agent/enable/", views.enable_email_agent), # email_agent:configure
path("<uuid:team_id>/email-agent/disable/", views.disable_email_agent),
]
```
- `POST /api/teams/{team_id}/email-agent` - Configure email agent for a team
- **Required Permissions:** Team access + `email_agent:configure`
- **Request Body:** `{ "inbox_email": "string", "ms_client_id": "string", "ms_client_secret": "string", "ms_tenant_id": "string" }`