Phase 1 fix auth redirect API handling and route issues
This commit is contained in:
@@ -61,7 +61,8 @@ def _redirect_login():
|
||||
|
||||
|
||||
def _redirect_denied():
|
||||
return RedirectResponse(url="/system-settings", status_code=303)
|
||||
from app.core.http_responses import ui_access_denied
|
||||
return ui_access_denied()
|
||||
|
||||
|
||||
def _flash_redirect(url: str) -> RedirectResponse:
|
||||
|
||||
@@ -22,7 +22,8 @@ def _redirect_login():
|
||||
|
||||
|
||||
def _redirect_denied():
|
||||
return RedirectResponse(url="/system-settings", status_code=303)
|
||||
from app.core.http_responses import ui_access_denied
|
||||
return ui_access_denied()
|
||||
|
||||
|
||||
def _is_system_admin(db, current_user) -> bool:
|
||||
|
||||
@@ -1,15 +1,58 @@
|
||||
from fastapi import APIRouter, Depends
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.core.db.deps import get_common_db
|
||||
from app.core.security.session_auth import require_login
|
||||
from app.modules.core.iam.models import User
|
||||
from app.modules.core.rbac.deps import get_user_roles
|
||||
from app.modules.core.tenancy.models import Tenant, Branch
|
||||
|
||||
router = APIRouter(prefix="/tenancy", tags=["tenancy"])
|
||||
|
||||
|
||||
def _require_tenancy_api_access(db: Session, user: User) -> None:
|
||||
roles = set(get_user_roles(db, int(user.id)))
|
||||
if not roles.intersection({"System Admin", "Firm Admin"}):
|
||||
raise HTTPException(status_code=403, detail="Tenancy API access denied")
|
||||
|
||||
|
||||
def _tenant_payload(t: Tenant) -> dict:
|
||||
return {
|
||||
"id": t.id,
|
||||
"code": getattr(t, "code", None),
|
||||
"name": getattr(t, "name", None),
|
||||
"is_active": getattr(t, "is_active", None),
|
||||
}
|
||||
|
||||
|
||||
def _branch_payload(b: Branch) -> dict:
|
||||
return {
|
||||
"id": b.id,
|
||||
"tenant_id": getattr(b, "tenant_id", None),
|
||||
"code": getattr(b, "code", None),
|
||||
"name": getattr(b, "name", None),
|
||||
"is_active": getattr(b, "is_active", None),
|
||||
}
|
||||
|
||||
|
||||
@router.get("/tenants")
|
||||
def list_tenants(db: Session = Depends(get_common_db)):
|
||||
return db.execute(select(Tenant).order_by(Tenant.id)).scalars().all()
|
||||
def list_tenants(db: Session = Depends(get_common_db), current_user: User = Depends(require_login)):
|
||||
_require_tenancy_api_access(db, current_user)
|
||||
stmt = select(Tenant).order_by(Tenant.id)
|
||||
roles = set(get_user_roles(db, int(current_user.id)))
|
||||
if "System Admin" not in roles:
|
||||
stmt = stmt.where(Tenant.id == current_user.tenant_id)
|
||||
return [_tenant_payload(t) for t in db.execute(stmt).scalars().all()]
|
||||
|
||||
|
||||
@router.get("/branches")
|
||||
def list_branches(db: Session = Depends(get_common_db)):
|
||||
return db.execute(select(Branch).order_by(Branch.id)).scalars().all()
|
||||
def list_branches(db: Session = Depends(get_common_db), current_user: User = Depends(require_login)):
|
||||
_require_tenancy_api_access(db, current_user)
|
||||
stmt = select(Branch).order_by(Branch.id)
|
||||
roles = set(get_user_roles(db, int(current_user.id)))
|
||||
if "System Admin" not in roles:
|
||||
stmt = stmt.where(Branch.tenant_id == current_user.tenant_id)
|
||||
return [_branch_payload(b) for b in db.execute(stmt).scalars().all()]
|
||||
|
||||
Reference in New Issue
Block a user