A R R R Associates
1014 lines
35 KiB
Python
1014 lines
35 KiB
Python
from __future__ import annotations
|
|
|
|
import secrets
|
|
from datetime import date, datetime, timezone
|
|
|
|
from fastapi import APIRouter, Form, Request
|
|
from fastapi.responses import JSONResponse, RedirectResponse
|
|
from sqlalchemy import select
|
|
|
|
from app.core.db.common import CommonSessionLocal
|
|
from app.core.security.csrf import get_or_create_csrf_token, validate_csrf
|
|
from app.core.security.passwords import hash_password
|
|
from app.core.security.session_auth import get_current_user
|
|
from app.core.templating import templates
|
|
from app.modules.core.audit.service import model_snapshot, pair_before_after, write_audit_log
|
|
from app.modules.core.iam.invite_service import issue_invite_token
|
|
from app.modules.email_integration.services import send_user_invite_email
|
|
from app.core.settings import get_settings
|
|
from app.modules.core.iam.lifecycle import (
|
|
LifecycleError,
|
|
activate_user,
|
|
deactivate_user,
|
|
disable_login,
|
|
enable_login,
|
|
ensure_manageable_lifecycle,
|
|
lock_user,
|
|
restore_user,
|
|
soft_delete_user,
|
|
unlock_user,
|
|
)
|
|
from app.modules.core.iam.models import User
|
|
from app.modules.core.iam.scope import (
|
|
build_scope,
|
|
ensure_assignable_roles,
|
|
ensure_manageable_existing_user,
|
|
ensure_users_manage_scope,
|
|
ensure_users_view_scope,
|
|
get_user_role_ids,
|
|
get_user_role_names,
|
|
list_scoped_users,
|
|
list_visible_branches,
|
|
list_visible_tenants,
|
|
resolve_target_tenant_branch,
|
|
)
|
|
from app.modules.core.iam.services import build_user_form_payload, build_user_listing_payload
|
|
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
|
|
from app.modules.core.rbac.models import UserRole
|
|
from app.modules.core.rbac.permission_guard import require_permission
|
|
from app.modules.core.tenancy.models import Branch
|
|
|
|
router = APIRouter(prefix="/system-settings/users", tags=["users-ui"])
|
|
|
|
|
|
def _public_invite_url(invite_token: str) -> str:
|
|
base = (get_settings().ERP_PUBLIC_BASE_URL or "").strip().rstrip("/") or "http://localhost:8000"
|
|
return f"{base}/invite/accept?token={invite_token}"
|
|
|
|
|
|
def _redirect_login():
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
|
|
def _redirect_denied():
|
|
from app.core.http_responses import ui_access_denied
|
|
return ui_access_denied()
|
|
|
|
|
|
def _flash_redirect(url: str) -> RedirectResponse:
|
|
return RedirectResponse(url=url, status_code=303)
|
|
|
|
|
|
def _common_context(request: Request, db, current_user: User, extra: dict | None = None):
|
|
ctx = {
|
|
"request": request,
|
|
"current_user": current_user,
|
|
"current_user_roles": get_user_roles(db, current_user.id),
|
|
"current_user_permissions": get_user_permissions(db, current_user.id),
|
|
"csrf_token": get_or_create_csrf_token(request),
|
|
}
|
|
if extra:
|
|
ctx.update(extra)
|
|
return ctx
|
|
|
|
|
|
def _user_admin_roles(db, current_user: User) -> set[str]:
|
|
return set(get_user_roles(db, current_user.id))
|
|
|
|
|
|
def _ensure_user_admin(db, current_user: User):
|
|
roles = _user_admin_roles(db, current_user)
|
|
if "System Admin" in roles or "Firm Admin" in roles:
|
|
return
|
|
raise PermissionError("Only System Admin and Firm Admin can create or manage users.")
|
|
|
|
|
|
INTERNAL_EMPLOYEE_ROLE_NAMES = {"Firm Admin", "Partner", "Manager", "Branch Manager", "Staff"}
|
|
EMPLOYEE_EXCLUDED_ROLE_NAMES = {"System Admin", "Client", "Consultant"}
|
|
|
|
|
|
def _role_name_set(roles) -> set[str]:
|
|
return {getattr(role, "name", "") for role in roles if getattr(role, "name", "")}
|
|
|
|
|
|
def _requires_employee_profile(role_names: set[str]) -> bool:
|
|
"""Internal firm users must have an employee profile.
|
|
|
|
Platform-only users, clients and consultants are intentionally excluded.
|
|
If an account has at least one internal employee role, we create/link the
|
|
employee profile, unless it is also a System Admin account.
|
|
"""
|
|
if "System Admin" in role_names:
|
|
return False
|
|
return bool(role_names.intersection(INTERNAL_EMPLOYEE_ROLE_NAMES))
|
|
|
|
|
|
def _parse_optional_date(value: str | None) -> date | None:
|
|
value = (value or "").strip()
|
|
if not value:
|
|
return None
|
|
return date.fromisoformat(value)
|
|
|
|
|
|
def _next_employee_code(db, tenant_id: int) -> str:
|
|
from app.modules.employees.models import Employee
|
|
|
|
prefix = "EMP"
|
|
rows = db.execute(
|
|
select(Employee.employee_code)
|
|
.where(Employee.tenant_id == tenant_id, Employee.employee_code.ilike(f"{prefix}%"))
|
|
.order_by(Employee.employee_code.desc())
|
|
).all()
|
|
max_no = 0
|
|
for (code,) in rows:
|
|
suffix = "".join(ch for ch in str(code or "") if ch.isdigit())
|
|
if suffix:
|
|
max_no = max(max_no, int(suffix))
|
|
return f"{prefix}{max_no + 1:05d}"
|
|
|
|
|
|
def _ensure_employee_profile_for_internal_user(
|
|
db,
|
|
*,
|
|
actor: User,
|
|
user_obj: User,
|
|
role_names: set[str],
|
|
employee_code: str | None = None,
|
|
mobile: str | None = None,
|
|
department: str | None = None,
|
|
designation: str | None = None,
|
|
date_of_joining: str | None = None,
|
|
employment_type: str | None = None,
|
|
) -> dict:
|
|
"""Create/link an Employee profile for internal firm users.
|
|
|
|
This function is intentionally additive. It does not create employee
|
|
profiles for System Admin, Client or Consultant-only users, and it does not
|
|
overwrite an existing linked employee profile.
|
|
"""
|
|
if not _requires_employee_profile(role_names):
|
|
return {"required": False, "created": False, "linked_existing": False, "employee_id": None}
|
|
|
|
if not user_obj.tenant_id or not user_obj.branch_id:
|
|
raise ValueError("Tenant and Branch are mandatory for internal employee users.")
|
|
|
|
from app.modules.employees.models import Employee
|
|
|
|
existing_linked = db.execute(
|
|
select(Employee).where(Employee.tenant_id == user_obj.tenant_id, Employee.user_id == user_obj.id)
|
|
).scalar_one_or_none()
|
|
if existing_linked:
|
|
return {"required": True, "created": False, "linked_existing": True, "employee_id": existing_linked.id}
|
|
|
|
email = (user_obj.email or "").strip().lower()
|
|
existing_unlinked = None
|
|
if email:
|
|
existing_unlinked = db.execute(
|
|
select(Employee).where(
|
|
Employee.tenant_id == user_obj.tenant_id,
|
|
Employee.user_id.is_(None),
|
|
Employee.email == email,
|
|
)
|
|
).scalar_one_or_none()
|
|
|
|
if existing_unlinked:
|
|
existing_unlinked.user_id = user_obj.id
|
|
existing_unlinked.branch_id = user_obj.branch_id
|
|
existing_unlinked.full_name = user_obj.full_name or existing_unlinked.full_name
|
|
existing_unlinked.updated_by_user_id = actor.id
|
|
existing_unlinked.updated_at_utc = datetime.now(timezone.utc)
|
|
return {"required": True, "created": False, "linked_existing": True, "employee_id": existing_unlinked.id}
|
|
|
|
code = (employee_code or "").strip() or _next_employee_code(db, int(user_obj.tenant_id))
|
|
code_exists = db.execute(
|
|
select(Employee).where(Employee.tenant_id == user_obj.tenant_id, Employee.employee_code == code)
|
|
).scalar_one_or_none()
|
|
if code_exists:
|
|
raise ValueError(f"Employee code '{code}' already exists in this tenant.")
|
|
|
|
emp_type = (employment_type or "full_time").strip() or "full_time"
|
|
if emp_type not in {"full_time", "part_time", "article_assistant", "intern", "consultant", "contract"}:
|
|
emp_type = "full_time"
|
|
|
|
employee = Employee(
|
|
tenant_id=int(user_obj.tenant_id),
|
|
branch_id=int(user_obj.branch_id),
|
|
user_id=user_obj.id,
|
|
employee_code=code,
|
|
full_name=(user_obj.full_name or email or code).strip(),
|
|
email=email or None,
|
|
mobile=(mobile or "").strip() or None,
|
|
date_of_joining=_parse_optional_date(date_of_joining),
|
|
employment_type=emp_type,
|
|
status="active",
|
|
is_active=True,
|
|
department=(department or "").strip() or None,
|
|
designation=(designation or "").strip() or None,
|
|
created_by_user_id=actor.id,
|
|
updated_by_user_id=actor.id,
|
|
)
|
|
db.add(employee)
|
|
db.flush()
|
|
return {"required": True, "created": True, "linked_existing": False, "employee_id": employee.id}
|
|
|
|
|
|
def _form_context(
|
|
db,
|
|
request: Request,
|
|
current_user: User,
|
|
scope,
|
|
*,
|
|
user_obj=None,
|
|
assigned_role_ids=None,
|
|
flash=None,
|
|
title="Create User",
|
|
form_mode="create",
|
|
):
|
|
payload = build_user_form_payload(db, scope, current_user, user_obj=user_obj, assigned_role_ids=assigned_role_ids)
|
|
payload.update({"title": title, "flash": flash, "form_mode": form_mode})
|
|
return _common_context(request, db, current_user, payload)
|
|
|
|
|
|
@router.get("")
|
|
def users_list(request: Request, q: str = "", page: int = 1, per_page: int = 10):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return _redirect_login()
|
|
|
|
try:
|
|
require_permission(db, current_user, "users.view")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
|
|
scope = build_scope(db, current_user)
|
|
try:
|
|
ensure_users_view_scope(scope)
|
|
except Exception:
|
|
return _flash_redirect("/system-settings")
|
|
|
|
payload = build_user_listing_payload(db, scope, q=q, page=page, per_page=per_page)
|
|
payload.update({"title": "Users", "scope": scope})
|
|
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/users_list.html",
|
|
_common_context(request, db, current_user, payload),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/branch-options")
|
|
def branch_options(request: Request, tenant_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return JSONResponse([], status_code=401)
|
|
|
|
try:
|
|
require_permission(db, current_user, "users.manage")
|
|
_ensure_user_admin(db, current_user)
|
|
except Exception:
|
|
return JSONResponse([], status_code=403)
|
|
|
|
scope = build_scope(db, current_user)
|
|
|
|
try:
|
|
ensure_users_manage_scope(scope)
|
|
except Exception:
|
|
return JSONResponse([], status_code=403)
|
|
|
|
if scope.is_system_admin:
|
|
branch_rows = db.execute(
|
|
select(Branch)
|
|
.where(Branch.tenant_id == tenant_id)
|
|
.order_by(Branch.name)
|
|
).scalars().all()
|
|
else:
|
|
if tenant_id != current_user.tenant_id:
|
|
return JSONResponse([], status_code=403)
|
|
branch_rows = list_visible_branches(db, scope, tenant_id)
|
|
|
|
return JSONResponse(
|
|
[
|
|
{"id": b.id, "name": b.name, "code": b.code}
|
|
for b in branch_rows
|
|
]
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/new")
|
|
def user_create(request: Request):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return _redirect_login()
|
|
|
|
try:
|
|
require_permission(db, current_user, "users.manage")
|
|
_ensure_user_admin(db, current_user)
|
|
except Exception:
|
|
return _redirect_denied()
|
|
|
|
scope = build_scope(db, current_user)
|
|
try:
|
|
ensure_users_manage_scope(scope)
|
|
except Exception:
|
|
return _flash_redirect("/system-settings/users")
|
|
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/user_form.html",
|
|
_form_context(db, request, current_user, scope, title="Create User", form_mode="create"),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/new")
|
|
def user_create_submit(
|
|
request: Request,
|
|
email: str = Form(...),
|
|
full_name: str = Form(...),
|
|
password: str = Form(...),
|
|
tenant_id: int | None = Form(None),
|
|
branch_id: int | None = Form(None),
|
|
role_ids: list[int] = Form([]),
|
|
is_active: str | None = Form(None),
|
|
allow_login: str | None = Form(None),
|
|
invite_user: str | None = Form(None),
|
|
employee_code: str = Form(""),
|
|
employee_mobile: str = Form(""),
|
|
employee_department: str = Form(""),
|
|
employee_designation: str = Form(""),
|
|
employee_date_of_joining: str = Form(""),
|
|
employee_employment_type: str = Form("full_time"),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return _redirect_login()
|
|
|
|
try:
|
|
require_permission(db, current_user, "users.manage")
|
|
_ensure_user_admin(db, current_user)
|
|
except Exception:
|
|
return _redirect_denied()
|
|
|
|
scope = build_scope(db, current_user)
|
|
|
|
try:
|
|
ensure_users_manage_scope(scope)
|
|
resolved_tenant_id, resolved_branch_id = resolve_target_tenant_branch(db, scope, tenant_id, branch_id)
|
|
roles = ensure_assignable_roles(db, scope, role_ids)
|
|
except Exception as exc:
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/user_form.html",
|
|
_form_context(
|
|
db,
|
|
request,
|
|
current_user,
|
|
scope,
|
|
assigned_role_ids=role_ids,
|
|
flash=str(exc),
|
|
title="Create User",
|
|
form_mode="create",
|
|
),
|
|
status_code=403,
|
|
)
|
|
|
|
email = email.lower().strip()
|
|
if db.execute(select(User).where(User.email == email)).scalar_one_or_none():
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/user_form.html",
|
|
_form_context(
|
|
db,
|
|
request,
|
|
current_user,
|
|
scope,
|
|
assigned_role_ids=role_ids,
|
|
flash="Email already exists.",
|
|
title="Create User",
|
|
form_mode="create",
|
|
),
|
|
status_code=400,
|
|
)
|
|
|
|
temp_password = password or secrets.token_urlsafe(12)
|
|
|
|
user = User(
|
|
email=email,
|
|
full_name=full_name.strip(),
|
|
password_hash=hash_password(temp_password),
|
|
tenant_id=resolved_tenant_id,
|
|
branch_id=resolved_branch_id,
|
|
is_active=is_active is not None,
|
|
allow_login=allow_login is not None,
|
|
is_locked=False,
|
|
deleted_at=None,
|
|
must_change_password=invite_user is not None,
|
|
password_changed_at_utc=None,
|
|
)
|
|
db.add(user)
|
|
db.flush()
|
|
|
|
for role in roles:
|
|
db.add(UserRole(user_id=user.id, role_id=role.id))
|
|
|
|
role_names = _role_name_set(roles)
|
|
try:
|
|
employee_link_result = _ensure_employee_profile_for_internal_user(
|
|
db,
|
|
actor=current_user,
|
|
user_obj=user,
|
|
role_names=role_names,
|
|
employee_code=employee_code,
|
|
mobile=employee_mobile,
|
|
department=employee_department,
|
|
designation=employee_designation,
|
|
date_of_joining=employee_date_of_joining,
|
|
employment_type=employee_employment_type,
|
|
)
|
|
except Exception as exc:
|
|
db.rollback()
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/user_form.html",
|
|
_form_context(
|
|
db,
|
|
request,
|
|
current_user,
|
|
scope,
|
|
assigned_role_ids=role_ids,
|
|
flash=f"User was not created: {exc}",
|
|
title="Create User",
|
|
form_mode="create",
|
|
),
|
|
status_code=400,
|
|
)
|
|
|
|
db.commit()
|
|
db.refresh(user)
|
|
|
|
invite_url = None
|
|
if invite_user is not None:
|
|
invite_token = issue_invite_token(db, user)
|
|
invite_url = _public_invite_url(invite_token)
|
|
|
|
write_audit_log(
|
|
db,
|
|
action="user.create",
|
|
entity_type="user",
|
|
actor=current_user,
|
|
request=request,
|
|
entity_id=user.id,
|
|
entity_name=user.email,
|
|
target_tenant_id=user.tenant_id,
|
|
target_branch_id=user.branch_id,
|
|
details={
|
|
"after": model_snapshot(
|
|
user,
|
|
["email", "full_name", "tenant_id", "branch_id", "is_active", "allow_login", "is_locked", "deleted_at"],
|
|
),
|
|
"role_ids": [role.id for role in roles],
|
|
"employee_profile": employee_link_result,
|
|
},
|
|
)
|
|
|
|
if invite_url:
|
|
try:
|
|
send_user_invite_email(db, user=user, invite_token=invite_token)
|
|
db.commit()
|
|
except Exception as exc:
|
|
print(f"[EMAIL INVITE ERROR] user={user.email} error={exc}")
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/invite_link.html",
|
|
_common_context(
|
|
request,
|
|
db,
|
|
current_user,
|
|
{"title": "Invite Link", "invite_url": invite_url, "invited_user": user},
|
|
),
|
|
)
|
|
|
|
return RedirectResponse(url="/system-settings/users", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{user_id}/invite")
|
|
def user_invite(request: Request, user_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return _redirect_login()
|
|
|
|
try:
|
|
require_permission(db, current_user, "users.manage")
|
|
require_permission(db, current_user, "users.invite")
|
|
_ensure_user_admin(db, current_user)
|
|
except Exception:
|
|
return _redirect_denied()
|
|
|
|
scope = build_scope(db, current_user)
|
|
user_obj = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user_obj:
|
|
return RedirectResponse(url="/system-settings/users", status_code=303)
|
|
|
|
ensure_manageable_existing_user(db, scope, user_obj)
|
|
|
|
invite_token = issue_invite_token(db, user_obj)
|
|
invite_url = _public_invite_url(invite_token)
|
|
try:
|
|
send_user_invite_email(db, user=user_obj, invite_token=invite_token)
|
|
db.commit()
|
|
except Exception as exc:
|
|
print(f"[EMAIL INVITE ERROR] user={user_obj.email} error={exc}")
|
|
|
|
write_audit_log(
|
|
db,
|
|
action="user.invited",
|
|
entity_type="user",
|
|
actor=current_user,
|
|
request=request,
|
|
entity_id=user_obj.id,
|
|
entity_name=user_obj.email,
|
|
target_tenant_id=user_obj.tenant_id,
|
|
target_branch_id=user_obj.branch_id,
|
|
details={"invite_url": invite_url},
|
|
)
|
|
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/invite_link.html",
|
|
_common_context(
|
|
request,
|
|
db,
|
|
current_user,
|
|
{"title": "Invite Link", "invite_url": invite_url, "invited_user": user_obj},
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/{user_id}/edit")
|
|
def user_edit(request: Request, user_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return _redirect_login()
|
|
|
|
try:
|
|
require_permission(db, current_user, "users.manage")
|
|
_ensure_user_admin(db, current_user)
|
|
except Exception:
|
|
return _redirect_denied()
|
|
|
|
scope = build_scope(db, current_user)
|
|
user_obj = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user_obj:
|
|
return RedirectResponse(url="/system-settings/users", status_code=303)
|
|
|
|
try:
|
|
ensure_manageable_existing_user(db, scope, user_obj)
|
|
ensure_users_manage_scope(scope)
|
|
except Exception:
|
|
return RedirectResponse(url="/system-settings/users", status_code=303)
|
|
|
|
assigned_role_ids = list(get_user_role_ids(db, user_id))
|
|
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/user_form.html",
|
|
_form_context(
|
|
db,
|
|
request,
|
|
current_user,
|
|
scope,
|
|
title=f"Edit User - {user_obj.email}",
|
|
form_mode="edit",
|
|
user_obj=user_obj,
|
|
assigned_role_ids=assigned_role_ids,
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{user_id}/edit")
|
|
def user_edit_submit(
|
|
request: Request,
|
|
user_id: int,
|
|
full_name: str = Form(...),
|
|
password: str = Form(""),
|
|
tenant_id: int | None = Form(None),
|
|
branch_id: int | None = Form(None),
|
|
role_ids: list[int] = Form([]),
|
|
is_active: str | None = Form(None),
|
|
allow_login: str | None = Form(None),
|
|
invite_user: str | None = Form(None),
|
|
employee_code: str = Form(""),
|
|
employee_mobile: str = Form(""),
|
|
employee_department: str = Form(""),
|
|
employee_designation: str = Form(""),
|
|
employee_date_of_joining: str = Form(""),
|
|
employee_employment_type: str = Form("full_time"),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return _redirect_login()
|
|
|
|
try:
|
|
require_permission(db, current_user, "users.manage")
|
|
_ensure_user_admin(db, current_user)
|
|
except Exception:
|
|
return _redirect_denied()
|
|
|
|
scope = build_scope(db, current_user)
|
|
user_obj = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user_obj:
|
|
return RedirectResponse(url="/system-settings/users", status_code=303)
|
|
|
|
try:
|
|
ensure_users_manage_scope(scope)
|
|
ensure_manageable_existing_user(db, scope, user_obj)
|
|
resolved_tenant_id, resolved_branch_id = resolve_target_tenant_branch(db, scope, tenant_id, branch_id)
|
|
roles = ensure_assignable_roles(db, scope, role_ids)
|
|
except Exception as exc:
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/user_form.html",
|
|
_form_context(
|
|
db,
|
|
request,
|
|
current_user,
|
|
scope,
|
|
title=f"Edit User - {user_obj.email}",
|
|
form_mode="edit",
|
|
user_obj=user_obj,
|
|
assigned_role_ids=role_ids,
|
|
flash=str(exc),
|
|
),
|
|
status_code=403,
|
|
)
|
|
|
|
before_snapshot = model_snapshot(
|
|
user_obj,
|
|
["email", "full_name", "tenant_id", "branch_id", "is_active", "allow_login", "is_locked", "deleted_at"],
|
|
)
|
|
old_role_ids = list(get_user_role_ids(db, user_obj.id))
|
|
|
|
user_obj.full_name = full_name.strip()
|
|
user_obj.tenant_id = resolved_tenant_id
|
|
user_obj.branch_id = resolved_branch_id
|
|
user_obj.is_active = is_active is not None
|
|
user_obj.allow_login = allow_login is not None
|
|
|
|
if password.strip():
|
|
user_obj.password_hash = hash_password(password.strip())
|
|
|
|
db.execute(UserRole.__table__.delete().where(UserRole.user_id == user_obj.id))
|
|
for role in roles:
|
|
db.add(UserRole(user_id=user_obj.id, role_id=role.id))
|
|
|
|
role_names = _role_name_set(roles)
|
|
try:
|
|
employee_link_result = _ensure_employee_profile_for_internal_user(
|
|
db,
|
|
actor=current_user,
|
|
user_obj=user_obj,
|
|
role_names=role_names,
|
|
employee_code=employee_code,
|
|
mobile=employee_mobile,
|
|
department=employee_department,
|
|
designation=employee_designation,
|
|
date_of_joining=employee_date_of_joining,
|
|
employment_type=employee_employment_type,
|
|
)
|
|
except Exception as exc:
|
|
db.rollback()
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/user_form.html",
|
|
_form_context(
|
|
db,
|
|
request,
|
|
current_user,
|
|
scope,
|
|
title=f"Edit User - {user_obj.email}",
|
|
form_mode="edit",
|
|
user_obj=user_obj,
|
|
assigned_role_ids=role_ids,
|
|
flash=f"User was not updated: {exc}",
|
|
),
|
|
status_code=400,
|
|
)
|
|
|
|
db.commit()
|
|
|
|
if invite_user is not None:
|
|
invite_token = issue_invite_token(db, user_obj)
|
|
invite_url = _public_invite_url(invite_token)
|
|
try:
|
|
send_user_invite_email(db, user=user_obj, invite_token=invite_token)
|
|
db.commit()
|
|
except Exception as exc:
|
|
print(f"[EMAIL INVITE ERROR] user={user_obj.email} error={exc}")
|
|
write_audit_log(
|
|
db,
|
|
action="user.invited",
|
|
entity_type="user",
|
|
actor=current_user,
|
|
request=request,
|
|
entity_id=user_obj.id,
|
|
entity_name=user_obj.email,
|
|
target_tenant_id=user_obj.tenant_id,
|
|
target_branch_id=user_obj.branch_id,
|
|
details={"invite_url": invite_url},
|
|
)
|
|
|
|
write_audit_log(
|
|
db,
|
|
action="user.update",
|
|
entity_type="user",
|
|
actor=current_user,
|
|
request=request,
|
|
entity_id=user_obj.id,
|
|
entity_name=user_obj.email,
|
|
target_tenant_id=user_obj.tenant_id,
|
|
target_branch_id=user_obj.branch_id,
|
|
details={
|
|
**pair_before_after(
|
|
before_snapshot,
|
|
model_snapshot(
|
|
user_obj,
|
|
["email", "full_name", "tenant_id", "branch_id", "is_active", "allow_login", "is_locked", "deleted_at"],
|
|
),
|
|
),
|
|
"old_role_ids": old_role_ids,
|
|
"new_role_ids": [role.id for role in roles],
|
|
"employee_profile": employee_link_result,
|
|
},
|
|
)
|
|
|
|
return RedirectResponse(url="/system-settings/users", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _lifecycle_template_response(request: Request, db, current_user: User, scope, flash: str, status_code: int = 400):
|
|
users = list_scoped_users(db, scope)
|
|
roles = {user_id: get_user_role_names(db, user_id) for user_id in [u.id for u in users]}
|
|
tenants = {t.id: t for t in list_visible_tenants(db, scope)}
|
|
|
|
visible_branch_tenant_id = current_user.tenant_id if not scope.is_system_admin else None
|
|
branch_rows = (
|
|
db.execute(select(Branch).order_by(Branch.name)).scalars().all()
|
|
if scope.is_system_admin
|
|
else list_visible_branches(db, scope, visible_branch_tenant_id)
|
|
)
|
|
branches = {b.id: b for b in branch_rows}
|
|
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/users_list.html",
|
|
_common_context(
|
|
request,
|
|
db,
|
|
current_user,
|
|
{
|
|
"title": "Users",
|
|
"users": users,
|
|
"user_roles": roles,
|
|
"tenants": tenants,
|
|
"branches": branches,
|
|
"scope": scope,
|
|
"flash": flash,
|
|
},
|
|
),
|
|
status_code=status_code,
|
|
)
|
|
|
|
|
|
def _run_lifecycle_action(db, request: Request, current_user: User, user_id: int, action: str):
|
|
scope = build_scope(db, current_user)
|
|
user_obj = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user_obj:
|
|
return RedirectResponse(url="/system-settings/users", status_code=303)
|
|
|
|
try:
|
|
_ensure_user_admin(db, current_user)
|
|
ensure_users_manage_scope(scope)
|
|
ensure_manageable_lifecycle(scope, db, current_user, user_obj, action.replace("-", " "))
|
|
except (Exception, LifecycleError) as exc:
|
|
return _lifecycle_template_response(request, db, current_user, scope, str(exc), 403)
|
|
|
|
before_snapshot = model_snapshot(
|
|
user_obj,
|
|
["email", "full_name", "tenant_id", "branch_id", "is_active", "allow_login", "is_locked", "deleted_at"],
|
|
)
|
|
|
|
if action == "activate":
|
|
activate_user(user_obj)
|
|
audit_action = "user.activate"
|
|
elif action == "deactivate":
|
|
deactivate_user(user_obj)
|
|
audit_action = "user.deactivate"
|
|
elif action == "enable-login":
|
|
enable_login(user_obj)
|
|
audit_action = "user.enable_login"
|
|
elif action == "disable-login":
|
|
disable_login(user_obj)
|
|
audit_action = "user.disable_login"
|
|
elif action == "lock":
|
|
lock_user(user_obj)
|
|
audit_action = "user.lock"
|
|
elif action == "unlock":
|
|
unlock_user(user_obj)
|
|
audit_action = "user.unlock"
|
|
elif action == "delete":
|
|
soft_delete_user(user_obj)
|
|
audit_action = "user.soft_delete"
|
|
elif action == "restore":
|
|
restore_user(user_obj)
|
|
audit_action = "user.restore"
|
|
else:
|
|
return _lifecycle_template_response(request, db, current_user, scope, "Unknown action", 400)
|
|
|
|
db.commit()
|
|
|
|
write_audit_log(
|
|
db,
|
|
action=audit_action,
|
|
entity_type="user",
|
|
actor=current_user,
|
|
request=request,
|
|
entity_id=user_obj.id,
|
|
entity_name=user_obj.email,
|
|
target_tenant_id=user_obj.tenant_id,
|
|
target_branch_id=user_obj.branch_id,
|
|
details=pair_before_after(
|
|
before_snapshot,
|
|
model_snapshot(
|
|
user_obj,
|
|
["email", "full_name", "tenant_id", "branch_id", "is_active", "allow_login", "is_locked", "deleted_at"],
|
|
),
|
|
),
|
|
)
|
|
|
|
return RedirectResponse(url="/system-settings/users", status_code=303)
|
|
|
|
|
|
@router.post("/{user_id}/activate")
|
|
def user_activate_submit(request: Request, user_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return _redirect_login()
|
|
try:
|
|
require_permission(db, current_user, "users.manage")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
return _run_lifecycle_action(db, request, current_user, user_id, "activate")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{user_id}/deactivate")
|
|
def user_deactivate_submit(request: Request, user_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return _redirect_login()
|
|
try:
|
|
require_permission(db, current_user, "users.manage")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
return _run_lifecycle_action(db, request, current_user, user_id, "deactivate")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{user_id}/enable-login")
|
|
def user_enable_login_submit(request: Request, user_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return _redirect_login()
|
|
try:
|
|
require_permission(db, current_user, "users.manage")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
return _run_lifecycle_action(db, request, current_user, user_id, "enable-login")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{user_id}/disable-login")
|
|
def user_disable_login_submit(request: Request, user_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return _redirect_login()
|
|
try:
|
|
require_permission(db, current_user, "users.manage")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
return _run_lifecycle_action(db, request, current_user, user_id, "disable-login")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{user_id}/lock")
|
|
def user_lock_submit(request: Request, user_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return _redirect_login()
|
|
try:
|
|
require_permission(db, current_user, "users.manage")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
return _run_lifecycle_action(db, request, current_user, user_id, "lock")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{user_id}/unlock")
|
|
def user_unlock_submit(request: Request, user_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return _redirect_login()
|
|
try:
|
|
require_permission(db, current_user, "users.manage")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
return _run_lifecycle_action(db, request, current_user, user_id, "unlock")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{user_id}/delete")
|
|
def user_delete_submit(request: Request, user_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return _redirect_login()
|
|
try:
|
|
require_permission(db, current_user, "users.manage")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
return _run_lifecycle_action(db, request, current_user, user_id, "delete")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{user_id}/restore")
|
|
def user_restore_submit(request: Request, user_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
current_user = get_current_user(request, db=db)
|
|
if not current_user:
|
|
return _redirect_login()
|
|
try:
|
|
require_permission(db, current_user, "users.manage")
|
|
except Exception:
|
|
return _redirect_denied()
|
|
return _run_lifecycle_action(db, request, current_user, user_id, "restore")
|
|
finally:
|
|
db.close() |