A R R R Associates
364 lines
14 KiB
Python
364 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, EmailStr
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.db.deps import get_common_db
|
|
from app.core.security.passwords import hash_password
|
|
from app.core.security.session_auth import require_login
|
|
from app.modules.core.audit.service import model_snapshot, pair_before_after, write_audit_log
|
|
from app.modules.core.iam.lifecycle import activate_user, deactivate_user, disable_login, enable_login, ensure_manageable_lifecycle, lock_user, restore_user, soft_delete_user, unlock_user, LifecycleError
|
|
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,
|
|
list_scoped_users,
|
|
resolve_target_tenant_branch,
|
|
scope_to_http,
|
|
)
|
|
from app.modules.core.rbac.deps import require_permission
|
|
from app.modules.core.rbac.models import UserRole
|
|
|
|
router = APIRouter(prefix="/users", tags=["users"])
|
|
|
|
|
|
class UserCreateRequest(BaseModel):
|
|
email: EmailStr
|
|
full_name: str
|
|
password: str
|
|
tenant_id: int | None = None
|
|
branch_id: int | None = None
|
|
role_ids: list[int] = []
|
|
is_active: bool = True
|
|
allow_login: bool = True
|
|
|
|
|
|
class UserUpdateRequest(BaseModel):
|
|
full_name: str
|
|
tenant_id: int | None = None
|
|
branch_id: int | None = None
|
|
role_ids: list[int] = []
|
|
is_active: bool = True
|
|
allow_login: bool = True
|
|
password: str | None = None
|
|
|
|
|
|
@router.get("", dependencies=[Depends(require_permission("users.view"))])
|
|
def list_users(current_user: User = Depends(require_login), db: Session = Depends(get_common_db)):
|
|
scope = build_scope(db, current_user)
|
|
try:
|
|
ensure_users_view_scope(scope)
|
|
except Exception as exc:
|
|
raise scope_to_http(exc)
|
|
|
|
users = list_scoped_users(db, scope)
|
|
return [
|
|
{
|
|
"id": u.id,
|
|
"email": u.email,
|
|
"full_name": u.full_name,
|
|
"tenant_id": u.tenant_id,
|
|
"branch_id": u.branch_id,
|
|
"is_active": u.is_active,
|
|
"allow_login": getattr(u, "allow_login", True),
|
|
"is_locked": getattr(u, "is_locked", False),
|
|
"deleted_at": (u.deleted_at.isoformat() if getattr(u, "deleted_at", None) else None),
|
|
}
|
|
for u in users
|
|
]
|
|
|
|
|
|
@router.post("", dependencies=[Depends(require_permission("users.manage"))])
|
|
def create_user(
|
|
payload: UserCreateRequest,
|
|
current_user: User = Depends(require_login),
|
|
db: Session = Depends(get_common_db),
|
|
):
|
|
scope = build_scope(db, current_user)
|
|
try:
|
|
ensure_users_manage_scope(scope)
|
|
tenant_id, branch_id = resolve_target_tenant_branch(db, scope, payload.tenant_id, payload.branch_id)
|
|
roles = ensure_assignable_roles(db, scope, payload.role_ids)
|
|
except Exception as exc:
|
|
raise scope_to_http(exc)
|
|
|
|
email = payload.email.lower().strip()
|
|
exists = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
|
|
if exists:
|
|
raise HTTPException(status_code=400, detail="Email already exists")
|
|
|
|
user = User(
|
|
email=email,
|
|
full_name=payload.full_name.strip(),
|
|
password_hash=hash_password(payload.password),
|
|
tenant_id=tenant_id,
|
|
branch_id=branch_id,
|
|
is_active=payload.is_active,
|
|
allow_login=payload.allow_login,
|
|
is_locked=False,
|
|
deleted_at=None,
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
|
|
for role in roles:
|
|
db.add(UserRole(user_id=user.id, role_id=role.id))
|
|
db.commit()
|
|
write_audit_log(
|
|
db,
|
|
action="user.create.api",
|
|
entity_type="user",
|
|
actor=current_user,
|
|
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]},
|
|
)
|
|
return {"status": "ok", "id": user.id}
|
|
|
|
|
|
@router.get("/{user_id}", dependencies=[Depends(require_permission("users.view"))])
|
|
def get_user(user_id: int, current_user: User = Depends(require_login), db: Session = Depends(get_common_db)):
|
|
scope = build_scope(db, current_user)
|
|
try:
|
|
ensure_users_view_scope(scope)
|
|
except Exception as exc:
|
|
raise scope_to_http(exc)
|
|
|
|
user = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
try:
|
|
ensure_manageable_existing_user(db, scope, user)
|
|
except Exception as exc:
|
|
raise scope_to_http(exc)
|
|
|
|
role_ids = db.execute(select(UserRole.role_id).where(UserRole.user_id == user.id)).scalars().all()
|
|
return {
|
|
"id": user.id,
|
|
"email": user.email,
|
|
"full_name": user.full_name,
|
|
"tenant_id": user.tenant_id,
|
|
"branch_id": user.branch_id,
|
|
"is_active": user.is_active,
|
|
"allow_login": getattr(user, "allow_login", True),
|
|
"is_locked": getattr(user, "is_locked", False),
|
|
"deleted_at": (user.deleted_at.isoformat() if getattr(user, "deleted_at", None) else None),
|
|
"role_ids": list(role_ids),
|
|
}
|
|
|
|
|
|
@router.put("/{user_id}", dependencies=[Depends(require_permission("users.manage"))])
|
|
def update_user(
|
|
user_id: int,
|
|
payload: UserUpdateRequest,
|
|
current_user: User = Depends(require_login),
|
|
db: Session = Depends(get_common_db),
|
|
):
|
|
scope = build_scope(db, current_user)
|
|
user = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
try:
|
|
ensure_users_manage_scope(scope)
|
|
ensure_manageable_existing_user(db, scope, user)
|
|
tenant_id, branch_id = resolve_target_tenant_branch(db, scope, payload.tenant_id, payload.branch_id)
|
|
roles = ensure_assignable_roles(db, scope, payload.role_ids)
|
|
except Exception as exc:
|
|
raise scope_to_http(exc)
|
|
|
|
before_snapshot = model_snapshot(user, ["email", "full_name", "tenant_id", "branch_id", "is_active", "allow_login", "is_locked", "deleted_at"])
|
|
old_role_ids = list(db.execute(select(UserRole.role_id).where(UserRole.user_id == user.id)).scalars().all())
|
|
|
|
user.full_name = payload.full_name.strip()
|
|
user.tenant_id = tenant_id
|
|
user.branch_id = branch_id
|
|
user.is_active = payload.is_active
|
|
user.allow_login = payload.allow_login
|
|
if payload.password:
|
|
user.password_hash = hash_password(payload.password)
|
|
|
|
db.execute(UserRole.__table__.delete().where(UserRole.user_id == user.id))
|
|
for role in roles:
|
|
db.add(UserRole(user_id=user.id, role_id=role.id))
|
|
db.commit()
|
|
write_audit_log(
|
|
db,
|
|
action="user.update.api",
|
|
entity_type="user",
|
|
actor=current_user,
|
|
entity_id=user.id,
|
|
entity_name=user.email,
|
|
target_tenant_id=user.tenant_id,
|
|
target_branch_id=user.branch_id,
|
|
details={**pair_before_after(before_snapshot, model_snapshot(user, ["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]},
|
|
)
|
|
return {"status": "ok"}
|
|
|
|
|
|
|
|
def _lifecycle_response(user: User) -> dict:
|
|
return {
|
|
"status": "ok",
|
|
"user_id": user.id,
|
|
"is_active": user.is_active,
|
|
"allow_login": getattr(user, "allow_login", True),
|
|
"is_locked": getattr(user, "is_locked", False),
|
|
"deleted_at": (user.deleted_at.isoformat() if getattr(user, "deleted_at", None) else None),
|
|
}
|
|
|
|
|
|
def _apply_lifecycle_action(db: Session, current_user: User, user: User, action: str):
|
|
before = model_snapshot(user, ["email", "full_name", "tenant_id", "branch_id", "is_active", "allow_login", "is_locked", "deleted_at"])
|
|
if action == "activate":
|
|
activate_user(user)
|
|
audit_action = "user.activate.api"
|
|
elif action == "deactivate":
|
|
deactivate_user(user)
|
|
audit_action = "user.deactivate.api"
|
|
elif action == "enable-login":
|
|
enable_login(user)
|
|
audit_action = "user.enable_login.api"
|
|
elif action == "disable-login":
|
|
disable_login(user)
|
|
audit_action = "user.disable_login.api"
|
|
elif action == "lock":
|
|
lock_user(user)
|
|
audit_action = "user.lock.api"
|
|
elif action == "unlock":
|
|
unlock_user(user)
|
|
audit_action = "user.unlock.api"
|
|
elif action == "delete":
|
|
soft_delete_user(user)
|
|
audit_action = "user.soft_delete.api"
|
|
elif action == "restore":
|
|
restore_user(user)
|
|
audit_action = "user.restore.api"
|
|
else:
|
|
raise HTTPException(status_code=400, detail="Unknown action")
|
|
db.commit()
|
|
write_audit_log(
|
|
db,
|
|
action=audit_action,
|
|
entity_type="user",
|
|
actor=current_user,
|
|
entity_id=user.id,
|
|
entity_name=user.email,
|
|
target_tenant_id=user.tenant_id,
|
|
target_branch_id=user.branch_id,
|
|
details=pair_before_after(before, model_snapshot(user, ["email", "full_name", "tenant_id", "branch_id", "is_active", "allow_login", "is_locked", "deleted_at"])),
|
|
)
|
|
return _lifecycle_response(user)
|
|
|
|
|
|
@router.post("/{user_id}/activate", dependencies=[Depends(require_permission("users.manage"))])
|
|
def activate_user_api(user_id: int, current_user: User = Depends(require_login), db: Session = Depends(get_common_db)):
|
|
scope = build_scope(db, current_user)
|
|
user = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
try:
|
|
ensure_manageable_lifecycle(scope, db, current_user, user, "activate")
|
|
except (Exception, LifecycleError) as exc:
|
|
raise scope_to_http(exc)
|
|
return _apply_lifecycle_action(db, current_user, user, "activate")
|
|
|
|
|
|
@router.post("/{user_id}/deactivate", dependencies=[Depends(require_permission("users.manage"))])
|
|
def deactivate_user_api(user_id: int, current_user: User = Depends(require_login), db: Session = Depends(get_common_db)):
|
|
scope = build_scope(db, current_user)
|
|
user = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
try:
|
|
ensure_manageable_lifecycle(scope, db, current_user, user, "deactivate")
|
|
except (Exception, LifecycleError) as exc:
|
|
raise scope_to_http(exc)
|
|
return _apply_lifecycle_action(db, current_user, user, "deactivate")
|
|
|
|
|
|
@router.post("/{user_id}/enable-login", dependencies=[Depends(require_permission("users.manage"))])
|
|
def enable_login_api(user_id: int, current_user: User = Depends(require_login), db: Session = Depends(get_common_db)):
|
|
scope = build_scope(db, current_user)
|
|
user = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
try:
|
|
ensure_manageable_lifecycle(scope, db, current_user, user, "enable login for")
|
|
except (Exception, LifecycleError) as exc:
|
|
raise scope_to_http(exc)
|
|
return _apply_lifecycle_action(db, current_user, user, "enable-login")
|
|
|
|
|
|
@router.post("/{user_id}/disable-login", dependencies=[Depends(require_permission("users.manage"))])
|
|
def disable_login_api(user_id: int, current_user: User = Depends(require_login), db: Session = Depends(get_common_db)):
|
|
scope = build_scope(db, current_user)
|
|
user = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
try:
|
|
ensure_manageable_lifecycle(scope, db, current_user, user, "disable login for")
|
|
except (Exception, LifecycleError) as exc:
|
|
raise scope_to_http(exc)
|
|
return _apply_lifecycle_action(db, current_user, user, "disable-login")
|
|
|
|
|
|
@router.post("/{user_id}/lock", dependencies=[Depends(require_permission("users.manage"))])
|
|
def lock_user_api(user_id: int, current_user: User = Depends(require_login), db: Session = Depends(get_common_db)):
|
|
scope = build_scope(db, current_user)
|
|
user = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
try:
|
|
ensure_manageable_lifecycle(scope, db, current_user, user, "lock")
|
|
except (Exception, LifecycleError) as exc:
|
|
raise scope_to_http(exc)
|
|
return _apply_lifecycle_action(db, current_user, user, "lock")
|
|
|
|
|
|
@router.post("/{user_id}/unlock", dependencies=[Depends(require_permission("users.manage"))])
|
|
def unlock_user_api(user_id: int, current_user: User = Depends(require_login), db: Session = Depends(get_common_db)):
|
|
scope = build_scope(db, current_user)
|
|
user = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
try:
|
|
ensure_manageable_lifecycle(scope, db, current_user, user, "unlock")
|
|
except (Exception, LifecycleError) as exc:
|
|
raise scope_to_http(exc)
|
|
return _apply_lifecycle_action(db, current_user, user, "unlock")
|
|
|
|
|
|
@router.post("/{user_id}/delete", dependencies=[Depends(require_permission("users.manage"))])
|
|
def delete_user_api(user_id: int, current_user: User = Depends(require_login), db: Session = Depends(get_common_db)):
|
|
scope = build_scope(db, current_user)
|
|
user = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
try:
|
|
ensure_manageable_lifecycle(scope, db, current_user, user, "delete")
|
|
except (Exception, LifecycleError) as exc:
|
|
raise scope_to_http(exc)
|
|
return _apply_lifecycle_action(db, current_user, user, "delete")
|
|
|
|
|
|
@router.post("/{user_id}/restore", dependencies=[Depends(require_permission("users.manage"))])
|
|
def restore_user_api(user_id: int, current_user: User = Depends(require_login), db: Session = Depends(get_common_db)):
|
|
scope = build_scope(db, current_user)
|
|
user = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
try:
|
|
ensure_manageable_lifecycle(scope, db, current_user, user, "restore")
|
|
except (Exception, LifecycleError) as exc:
|
|
raise scope_to_http(exc)
|
|
return _apply_lifecycle_action(db, current_user, user, "restore")
|