146 lines
5.2 KiB
Python
146 lines
5.2 KiB
Python
import uuid
|
|
from typing import List, Optional
|
|
from fastapi import APIRouter, Depends, status, HTTPException, Query, Request
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config.database import get_db
|
|
from app.controllers.auth.tenant_controller import TenantController
|
|
from app.schemas.auth.tenant_schema import TenantCreate, TenantUpdate, TenantResponse, TenantPaginatedResponse
|
|
from app.middleware.auth_middleware import get_current_user, require_access, User
|
|
from app.middleware.tenant_middleware import get_tenant_id_from_user
|
|
from app.services.system.audit_log_service import AuditLogService
|
|
from app.helper.helpers import get_client_ip
|
|
|
|
router = APIRouter()
|
|
|
|
@router.post("/create", response_model=TenantResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_tenant(
|
|
request: Request,
|
|
tenant_data: TenantCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
_ = Depends(require_access("superadmin.tenant.create"))
|
|
):
|
|
result = TenantController.create_tenant(db, tenant_data)
|
|
|
|
AuditLogService.log(
|
|
db=db,
|
|
module_name="Tenants",
|
|
action_type="CREATE",
|
|
entity_id=str(result.id),
|
|
entity_name=result.tenant_name,
|
|
description=f"Tenant '{result.tenant_name}' onboarded",
|
|
performed_by_id=str(current_user.id),
|
|
performed_by_email=current_user.email,
|
|
ip_address=_get_client_ip(request),
|
|
new_values=tenant_data.model_dump(mode='json')
|
|
)
|
|
return result
|
|
|
|
@router.get("/get", response_model=List[TenantResponse])
|
|
def get_all_tenants(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
_ = Depends(require_access("superadmin.tenant.read"))
|
|
):
|
|
# READ actions are typically not logged to avoid DB bloat
|
|
return TenantController.get_all_tenants(db)
|
|
|
|
@router.get("/me", response_model=TenantResponse)
|
|
def get_my_tenant(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
tenant_id = get_tenant_id_from_user(current_user)
|
|
if not tenant_id:
|
|
raise HTTPException(status_code=404, detail="Tenant not found")
|
|
return TenantController.get_tenant_by_id(db, tenant_id)
|
|
|
|
@router.put("/update/{tenant_id}", response_model=TenantResponse)
|
|
def update_tenant(
|
|
request: Request,
|
|
tenant_id: uuid.UUID,
|
|
tenant_data: TenantUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
_ = Depends(require_access("superadmin.tenant.update"))
|
|
):
|
|
# 1. Get snapshot BEFORE update for Audit Log
|
|
existing = TenantController.get_tenant_by_id(db, tenant_id)
|
|
old_snapshot = {
|
|
"tenant_name": existing.tenant_name,
|
|
"tenant_domain": existing.tenant_domain,
|
|
"is_active": existing.is_active
|
|
}
|
|
|
|
# 2. Perform update
|
|
result = TenantController.update_tenant(db, tenant_id, tenant_data)
|
|
|
|
# 3. Prepare new values and indentify deltas
|
|
new_snapshot = tenant_data.model_dump(mode='json', exclude_unset=True)
|
|
|
|
old_values = {k: old_snapshot[k] for k in new_snapshot if k in old_snapshot and old_snapshot[k] != new_snapshot[k]}
|
|
new_values = {k: new_snapshot[k] for k in old_values}
|
|
|
|
# 4. Log the change
|
|
AuditLogService.log(
|
|
db=db,
|
|
module_name="Tenants",
|
|
action_type="UPDATE",
|
|
entity_id=str(tenant_id),
|
|
entity_name=result.tenant_name,
|
|
description=f"Tenant '{result.tenant_name}' updated",
|
|
performed_by_id=str(current_user.id),
|
|
performed_by_email=current_user.email,
|
|
ip_address=_get_client_ip(request),
|
|
old_values=old_values,
|
|
new_values=new_values
|
|
)
|
|
return result
|
|
|
|
@router.delete("/delete/{tenant_id}")
|
|
def delete_tenant(
|
|
request: Request,
|
|
tenant_id: uuid.UUID,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
_ = Depends(require_access("superadmin.tenant.delete"))
|
|
):
|
|
# 1. Get snapshot before deletion
|
|
existing = TenantController.get_tenant_by_id(db, tenant_id)
|
|
|
|
# 2. Perform deletion
|
|
result = TenantController.delete_tenant(db, tenant_id)
|
|
|
|
# 3. Log the deletion
|
|
AuditLogService.log(
|
|
db=db,
|
|
module_name="Tenants",
|
|
action_type="DELETE",
|
|
entity_id=str(tenant_id),
|
|
entity_name=existing.tenant_name,
|
|
description=f"Tenant '{existing.tenant_name}' removed",
|
|
performed_by_id=str(current_user.id),
|
|
performed_by_email=current_user.email,
|
|
ip_address=_get_client_ip(request),
|
|
old_values={"tenant_name": existing.tenant_name, "domain": existing.tenant_domain}
|
|
)
|
|
return result
|
|
|
|
@router.get("/list", response_model=TenantPaginatedResponse)
|
|
def list_tenants(
|
|
page: int = Query(1, ge=1, description="Page number"),
|
|
page_size: int = Query(10, ge=1, le=100, description="Items per page"),
|
|
search: Optional[str] = Query(None, description="Search in name, domain, or ID"),
|
|
is_active: Optional[bool] = Query(None, description="Filter by active status"),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
_ = Depends(require_access("superadmin.tenant.read")),
|
|
):
|
|
return TenantController.get_tenants_paginated(
|
|
db=db,
|
|
page=page,
|
|
page_size=page_size,
|
|
search=search,
|
|
is_active=is_active,
|
|
) |